{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 第4章 构建低延迟应用的C++基础模块"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "在上一章，我们对如何使用C++开发低延迟应用进行了详细且深入的技术探讨。我们还研究了C++编程语言以及GCC编译器的技术细节。现在，我们将从理论探讨转向自己动手构建一些实用的低延迟C++组件。\n",
    "\n",
    "我们将构建一些相对通用的组件，这些组件可用于多种不同的低延迟应用，就像我们在前几章讨论过的那些应用一样。在本章构建这些基础模块的过程中，我们将学习如何有效地使用C++编写高性能的C++代码。在本书的后续部分，我们会使用这些组件来展示它们如何融入我们即将设计和构建的电子交易生态系统。\n",
    "\n",
    "在本章中，我们将涵盖以下主题：\n",
    "\n",
    "- 用于多线程低延迟应用的C++线程机制\n",
    "- 设计C++内存池以避免动态内存分配\n",
    "- 使用无锁队列传输数据\n",
    "- 构建低延迟日志框架\n",
    "- 使用套接字进行C++网络编程"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 技术要求"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "本章的源代码位于仓库中的[`Chapter4`](https://github.com/PacktPublishing/Building-Low-Latency-Applications-with-CPP/tree/main/Chapter4)目录下。\n",
    "\n",
    "我们期望你至少具备中级C++编程经验，因为我们假定你对广泛使用的C++编程特性有很好的理解。我们还假定你有一些C++网络编程经验，由于网络编程是一个庞大的主题，本书无法涵盖。从本章开始，本书将使用CMake和Ninja构建系统，所以我们希望你了解CMake、g++、Ninja、Make或类似的构建系统，以便能够编译本书的代码示例。\n",
    "\n",
    "此处展示了本书源代码开发环境的配置信息。我们提供此环境的详细信息，是因为本书中给出的所有C++代码不一定具有可移植性，可能需要进行一些小的修改才能在你的环境中运行：\n",
    "\n",
    "- 操作系统（OS）：Linux 5.19.0 - 41 - generic #42~22.04.1 - Ubuntu SMP PREEMPT_DYNAMIC Tue Apr 18 17:40:00 UTC 2 x86_64 x86_64 x86_64 GNU/Linux\n",
    "- GCC：g++ (Ubuntu 11.3.0 - 1ubuntu1~22.04.1) 11.3.0\n",
    "- CMake：cmake version 3.23.2\n",
    "- Ninja：1.10.2"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 用于多线程低延迟应用的C++线程机制"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "我们要构建的第一个组件虽然很小，但却非常基础。本节将设计并实现一种创建和运行执行线程的方法。根据系统中不同子组件的设计，这些线程会在完整的低延迟系统的许多不同部分中使用。根据系统设计，不同组件可能会像流水线一样协同工作，以实现并行处理。在我们的电子交易系统中，我们正是以这种方式使用多线程框架的。另一个用例是将非关键任务，如将日志记录到磁盘、计算统计信息等，交给后台线程处理。\n",
    "\n",
    "在深入研究创建和操作线程的源代码之前，让我们先快速定义几个有用的宏。从本章开始，在我们编写的源代码中，很多地方都会用到这些函数。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 定义一些有用的宏和函数"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "大多数低延迟应用运行在现代流水线处理器上，这些处理器会在指令和数据需要执行之前进行预取。我们在上一章讨论过，分支预测错误的代价非常高，会导致流水线停顿，产生空闲周期。因此，对于低延迟应用来说，一个重要的开发实践是减少分支。由于分支不可避免，所以尽量使它们具有可预测性也很重要。\n",
    "\n",
    "我们有两个简单的宏，用于向编译器提供分支提示。它们使用了GCC内置函数`builtin_expect`，该函数会对编译器生成的机器指令进行重新排序。实际上，编译器会根据开发者提供的分支预测提示，在假设某个分支更有可能或更不可能被执行的情况下生成优化的机器代码。\n",
    "\n",
    "请注意，指令重排序只是分支预测的一部分，因为处理器在运行指令时还会使用硬件分支预测器。现代硬件分支预测器在预测分支和跳转方面非常出色，尤其是在同一分支多次被执行，甚至存在容易预测的分支模式的情况下。\n",
    "\n",
    "这两个宏如下："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "#define LIKELY(x)    builtin_expect(!!(x), 1)\n",
    "#define UNLIKELY(x)    builtin_expect(!!(x), 0)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`LIKELY(x)`宏指定由`x`指定的条件很可能为真，而`UNLIKELY(x)`宏的作用则相反。作为使用示例，我们将在下一组函数中很快使用`UNLIKELY`宏。在C++20中，这一功能被标准化为`[[likely]]`和`[[unlikely]]`属性，以一种标准且可移植的方式实现相同的功能。\n",
    "\n",
    "接下来我们将定义另外两个函数，它们仅用于我们代码库中的断言。这些函数应该很容易理解；`ASSERT`会在其传入的条件为假时记录一条消息并退出程序，`FATAL`则只是记录一条消息并退出程序。注意这里使用`UNLIKELY`来指定我们不期望`!cond`条件为真。还要注意，在关键代码路径上使用`ASSERT`方法并非没有开销，主要是因为其中的`if`检查。在发布版本的代码中，我们最终会对其进行优化，将其从代码中去除，但目前我们先保留它，因为它的使用成本应该非常低。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "inline auto ASSERT(bool cond, const std::string& msg)\n",
    "noexcept {\n",
    "    if(UNLIKELY(!cond)) {\n",
    "        std::cerr << msg << std::endl;\n",
    "        exit(EXIT_FAILURE);\n",
    "    }\n",
    "}\n",
    "\n",
    "inline auto FATAL(const std::string& msg) noexcept {\n",
    "    std::cerr << msg << std::endl;\n",
    "    exit(EXIT_FAILURE);\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "本节讨论的代码可以在本书GitHub仓库中的[`Chapter4/macros.h`](https://github.com/PacktPublishing/Building-Low-Latency-Applications-with-CPP/blob/main/Chapter4/macros.h)源文件中找到。注意，`macros.h`头文件包含以下两个头文件："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "#include <cstring>\n",
    "#include <iostream>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "现在，让我们在下一节深入研究线程创建和操作功能。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 创建并启动新线程\n",
    "\n",
    "以下代码块中定义的方法创建一个新的线程对象，设置线程的亲和力（稍后会详细介绍），并转发线程执行期间将运行的函数及相关参数。这是通过`thread_body` lambda表达式实现的，该表达式被传递给`std::thread`的构造函数。注意这里使用了可变参数模板和完美转发，使得这个方法可以运行各种函数、任意类型的参数，并且参数数量不受限制。创建线程后，该方法会等待，直到线程成功启动或因设置线程亲和力失败而启动失败，这就是调用`t->join()`的作用。目前先忽略对`setThreadCore(core_id)`的调用，我们将在下一节讨论它。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- [thread_utils.h](./libcommon/thread_utils.hpp)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "本节讨论的代码可以在本书GitHub仓库中的[`Chapter4/thread_utils.h`](https://github.com/PacktPublishing/Building-Low-Latency-Applications-with-CPP/blob/main/Chapter4/thread_utils.h)源文件中找到。现在，让我们进入最后一节，了解`setThreadCore(core_id)`函数中设置线程亲和力的相关内容。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 设置线程亲和力\n",
    "\n",
    "在这里，我们将讨论为上一节中线程创建lambda表达式设置线程亲和力的源代码。在讨论源代码之前，请记住，如果线程之间频繁进行上下文切换，会给线程性能带来很大开销。线程在不同CPU核心之间切换也会因类似原因影响性能。对于低延迟应用来说，为性能关键型线程设置线程亲和力非常重要，这样可以避免这些问题。\n",
    "\n",
    "现在，让我们看看在`setThreadCore()`方法中如何设置线程亲和力。首先，我们使用`CPU_ZERO()`方法清空`cpu_set_t`变量，它实际上就是一个标志数组。然后，我们使用`CPU_SET()`方法为要绑定的核心ID启用相应的条目。最后，我们使用`pthread_setaffinity_np()`函数设置线程亲和力，如果设置失败则返回`false`。注意这里使用`pthread_self()`来获取要使用的线程ID，这是合理的，因为这个函数是在`createAndStartThread()`中创建的`std::thread`实例内部被调用的。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- [thread_utils.h](./libcommon/thread_utils.hpp)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "本节讨论的代码可以在本书GitHub仓库中的[`Chapter4/thread_utils.h`](https://github.com/PacktPublishing/Building-Low-Latency-Applications-with-CPP/blob/main/Chapter4/thread_utils.h)源文件中找到。当你查看GitHub仓库中的`thread_utils.h`源文件时会发现，这些代码块属于`Common`命名空间。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 构建一个示例"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "在结束本节内容之前，让我们快速看一个使用刚刚创建的线程工具的简单示例。该示例位于本书GitHub仓库中`Chapter4/thread_example.cpp`源文件里。请注意，本章的库以及所有示例都可以使用`Chapter4`目录下的`CMakeLists.txt`进行构建。我们还提供了两个简单的脚本`build.sh`和`run_examples.sh`，在设置好`cmake`和`ninja`二进制文件的正确路径后，可用于构建和运行这些示例。这里要注意，`cmake`和`ninja`只是随意选择的构建系统，如果有需要，你可以将构建系统更改为其他的。\n",
    "\n",
    "这个示例应该很好理解——我们创建并启动两个线程，每个线程执行一个虚拟任务，即把传入的两个参数（`a`和`b`）相加。然后，在程序退出前，我们等待线程执行完毕："
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- [thread_example.cpp](./thread_example/thread_example.cpp)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "! cd ../../ && rm -rf build && cmake -S . -B build && cmake --build build --target thread_example --config Release "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Set core affinity for dummyFunction1 129537513879104 to -1\n",
      "dummyFunction(12,21)\n",
      "dummyFunction output=33\n",
      "dummyFunction done.\n",
      "Set core affinity for dummyFunction2 129537505486400 to 1\n",
      "dummyFunction(15,51)\n",
      "dummyFunction output=66\n",
      "dummyFunction sleeping...\n",
      "Set core affinity for dummyFunction3 129537421604416 to 5\n",
      "dummyFunction(1,2)\n",
      "dummyFunction output=3\n",
      "dummyFunction sleeping...\n",
      "Set core affinity for dummyFunction4 129537413211712 to 8\n",
      "dummyFunction(3,4)\n",
      "dummyFunction output=7\n",
      "dummyFunction sleeping...\n",
      "Set core affinity for dummyFunction5 129537404819008 to 15\n",
      "dummyFunction(5,6)\n",
      "dummyFunction output=11\n",
      "dummyFunction sleeping...\n",
      "Set core affinity for dummyFunction6 129537396426304 to 0\n",
      "dummyFunction(7,8)\n",
      "dummyFunction output=15\n",
      "dummyFunction sleeping...\n",
      "dummyFunction done.\n",
      "main waiting for threads to be done.\n",
      "dummyFunction done.\n",
      "dummyFunction done.\n",
      "dummyFunction done.\n",
      "dummyFunction done.\n",
      "main exiting.\n"
     ]
    }
   ],
   "source": [
    "! cd ../../build/Part_1/04/thread_example && ./thread_example"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "接下来进入下一节，我们将讨论在运行时需要创建和丢弃对象的情况下，如何避免**动态内存分配（dynamic memory allocation）**。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 设计C++内存池以避免动态内存分配\n",
    "\n",
    "我们已经多次讨论过动态内存分配、操作系统执行动态内存分配的步骤，以及为什么动态内存分配速度很慢。实际上，动态内存分配速度非常慢，以至于低延迟应用程序会尽可能在关键路径上避免使用它。在运行时，我们构建有用的应用程序离不开创建和删除大量对象，然而对于低延迟应用程序来说，动态内存分配的速度实在太慢。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 理解内存池的定义\n",
    "\n",
    "首先，让我们正式定义一下什么是**内存池（memory pool）**，以及为什么需要它。许多应用程序（包括低延迟应用程序）需要能够处理大量对象，而且对象的数量是未知的。这里所说的对象数量未知，指的是无法提前确定预期的对象数量，也无法确定对象的最大数量是多少。显然，可能的最大对象数量受限于系统内存的容纳量。处理这些对象的传统方法是根据需要使用动态内存分配。在这种情况下，**堆内存（heap memory）**被视为内存池，也就是用于分配和释放内存的内存池。遗憾的是，这种方式速度很慢，因此我们将在自己的系统中使用自定义的内存池来控制内存的分配和释放过程。我们将内存池定义为可以从中请求额外内存或对象，并可以将空闲内存或对象返回的地方。通过构建自定义的内存池，我们可以利用使用模式，控制分配和释放机制，以实现最佳性能。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 理解内存池的使用场景\n",
    "\n",
    "当提前知道需要某种特定类型对象的确切数量时，你可以选择在需要时精确创建相应数量的对象。但在实际情况中，很多时候无法提前得知对象的确切数量。这意味着我们需要使用动态内存分配即时创建对象。如前所述，动态内存分配是一个非常缓慢的过程，对于低延迟应用程序来说是个问题。我们用“内存池”这个术语来描述某种类型对象的集合，这也是我们在本节要构建的内容。在本书中，我们将使用内存池来分配和释放那些数量无法预测的对象。\n",
    "\n",
    "我们采用的解决方案是在启动时预先分配大量内存块，然后在运行时根据需要分配相应数量的内存，也就是说，我们自己从这个存储池中执行内存的分配和释放操作。这样做在很多方面表现得更好，例如，我们可以将内存池的使用限制在系统的特定组件中，而不是服务器上运行的所有进程。我们还可以控制内存存储、分配和释放算法，根据特定应用进行调整，以实现最佳性能。\n",
    "\n",
    "让我们首先为内存池做出一些设计决策。内存池的所有源代码位于本书GitHub仓库中的[`Chapter4/mem_pool.h`](https://github.com/PacktPublishing/Building-Low-Latency-Applications-with-CPP/blob/main/Chapter4/mem_pool.h)源文件里。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 设计内存池存储"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "首先，我们需要确定如何在内存池中存储元素。这里主要有两种选择：使用类似老式数组（`T[N]`）或`std::array`在栈上存储，或者使用类似老式指针（`T*`）或`std::vector`在堆上存储。根据内存池的大小、使用频率、使用模式以及应用程序本身的情况，一种选择可能会优于另一种。例如，如果我们预计内存池需要大量内存，可能是因为存储的对象很大，或者对象数量很多，在这种情况下，堆分配会是更好的选择，这样可以满足大内存需求，同时不会影响栈内存。如果预计对象数量很少或者对象很小，我们应该考虑使用栈实现。如果预计很少访问对象，将它们存储在栈上可能会获得更好的缓存性能，但对于频繁访问的情况，两种实现方式效果可能差不多。和许多其他选择一样，这些决策通常要通过实际测量性能来确定。对于我们的内存池，我们将使用`std::vector`和堆分配，不过要注意这不是线程安全的。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "我们还需要一个变量来跟踪哪些内存块是空闲的，哪些是正在使用的。最后，还需要一个变量来跟踪下一个空闲内存块的位置，以便快速处理分配请求。这里有一点很重要，我们有两种选择：\n",
    "\n",
    "- 我们使用两个向量，一个用于跟踪对象，另一个用于跟踪空闲或已占用标记。下面的图表展示了这种解决方案；请注意，在这个示例中，我们假设这两个向量位于截然不同的内存位置。这里要说明的是，访问空闲或已占用标记以及对象本身可能会导致缓存未命中（cache misses），因为它们在内存中彼此相距较远。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"./images/4.1.png\" alt=\"Figure 4.1 – A memory pool implementation that uses two vectors to track objects and show which indices are free or in use\" style=\"zoom:50%;\" />\n",
    "\n",
    "图4.1 使用两个向量跟踪对象并显示哪些索引空闲或已占用的内存池实现"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 我们维护一个结构体（可以是结构体、类或基本类型对象）的单一向量，每个结构体同时存储对象和表示空闲或已占用标志的变量。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"./images/4.2.png\" alt=\"Figure 4.2 – A memory pool implementation that uses a single vector to track the object and see whether it is free or in use\" style=\"zoom:50%;\" />\n",
    "\n",
    "图4.2 使用单个向量跟踪对象并判断其是否空闲的内存池实现"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "从缓存性能的角度来看，第二种选择更好，因为访问对象和紧跟在对象后面的空闲标记，比访问两个不同向量中可能在内存中相距较远的不同位置要好。这也是因为在几乎所有的使用模式中，如果我们访问对象，通常也会访问空闲标记，反之亦然。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "/**\n",
    " * @brief 内部存储结构：封装 T 对象和空闲标记。\n",
    " *\n",
    " * 将这两个信息放在一个结构体中有助于缓存局部性，因为分配/释放时通常会同时访问两者。\n",
    " */\n",
    "struct ObjectBlock {\n",
    "  T object_;       ///< 预留给对象的存储空间（生命周期由 placement new 控制）\n",
    "  bool is_free_ = true; ///< 当前槽位是否空闲\n",
    "};"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "接下来，我们需要研究如何在构造函数中初始化这个内存池，以及一些用于构造和赋值任务的样板代码。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 初始化内存池\n",
    "\n",
    "初始化内存池非常简单直接——我们只需要接受一个参数，该参数指定内存池的初始大小，然后将向量初始化为足够大，以容纳那么多同时分配的对象。在我们的设计中，不会添加调整内存池大小超过其初始大小的功能，但如果有需要，这是一个相对简单的扩展。请注意，这个初始向量的初始化是内存池唯一一次动态分配内存，所以内存池应该在关键路径执行之前创建。这里需要注意的一点是，我们添加了一个断言，以确保`T`类型的实际对象是`ObjectBlock`结构体中的第一个成员；在“处理内存释放”部分，我们将看到这样做的原因："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "/**\n",
    " * @brief 构造固定容量的内存池。\n",
    " *\n",
    " * @param num_elems 预分配的槽位数量。\n",
    " * @note 这里使用 vector<ObjectBlock>\n",
    " * 进行存储，以确保动态分配，避免栈上大块内存。\n",
    " */\n",
    "explicit MemPool(std::size_t num_elems)\n",
    "    : store_(num_elems, {T(), true}) /* 预先构造 num_elems 个 ObjectBlock，默认标记为空闲。 */\n",
    "{\n",
    "    // 断言：确保 ObjectBlock 内部的 T 对象与结构体起始地址对齐。\n",
    "    // 若失败，说明 T 不是首个成员，后续将 T* 转换为 ObjectBlock*\n",
    "    // 的做法不再安全。\n",
    "    ASSERT(reinterpret_cast<const ObjectBlock*>(&(store_[0].object_)) ==\n",
    "               &(store_[0]),\n",
    "           \"T object should be first member of ObjectBlock.\");\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "现在来看一些模板代码——我们将删除默认构造函数、复制构造函数和移动构造函数。对于复制赋值运算符和移动赋值运算符，我们也会进行同样的操作。这样做是为了防止在我们不知情的情况下意外调用这些方法。这也是我们将构造函数声明为`explicit`的原因——以禁止我们不期望的隐式转换："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// 禁用默认构造、拷贝/移动构造与赋值，避免无意间复制内存池状态。\n",
    "MemPool() = delete;\n",
    "MemPool(const MemPool &) = delete;\n",
    "MemPool(const MemPool &&) = delete;\n",
    "MemPool &operator=(const MemPool &) = delete;\n",
    "MemPool &operator=(const MemPool &&) = delete;"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "现在，让我们继续编写代码，通过提供一个`T`类型模板参数的空闲对象来满足分配请求。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 处理新的分配请求"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "处理分配请求就是在内存池存储中找到一个空闲块，这我们可以通过`next_free_index_`跟踪器轻松完成。然后，我们更新该块的`is_free_`标记，使用定位`new`初始化`T`类型的对象块，然后更新`next_free_index_`，使其指向下一个可用的空闲块。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "注意两点\n",
    "- 第一点是，我们使用定位`new`返回一个`T`类型的对象，而不是一个与`T`大小相同的内存块。这并不是绝对必要的，如果内存池的用户希望自己从我们返回的内存块构造对象，可以去掉这一步。在大多数编译器实现中，定位`new`可能会添加一个额外的`if`检查，以确认提供给它的内存块不为空。\n",
    "- 第二点，这更多是根据应用场景做出的设计选择，我们调用`updateNextFreeIndex()`来更新`next_free_index_`，使其指向下一个可用的空闲块，除了这里提供的实现方式，还有其他不同的实现方式。至于哪种实现是最优的，这取决于具体情况，需要在实践中进行测试。现在，让我们先看看`allocate()`方法，在这里我们再次使用可变参数模板，以便将任意参数转发到`T`的构造函数。注意，这里我们使用定位`new`操作符，从内存块中使用给定参数构造一个`T`类型的对象。记住，`new`是一个操作符，如果需要也可以被重写，而定位`new`操作符会跳过分配内存的步骤，直接使用提供的内存块：\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "/**\n",
    " * @brief 在内存池中分配一个对象，并调用 T 的构造函数。\n",
    " *\n",
    " * @tparam Args 完美转发给 T 构造函数的参数类型。\n",
    " * @param args  构造 T 所需的参数，以按值传递方式接收。\n",
    " * @return 指向新构造对象的指针。\n",
    " *\n",
    " * @note\n",
    " *  - 使用 placement new 在预先分配的内存上构造对象。\n",
    " *  - 每次分配都会更新 `next_free_index_` 指向下一个可用槽位。\n",
    " */\n",
    "template <typename... Args>\n",
    "T* allocate(Args... args) noexcept {\n",
    "    auto obj_block = &(store_[next_free_index_]);  // 获取当前空闲槽位\n",
    "    // 断言：确认当前索引对应的块确实空闲，避免覆盖正在使用的对象。\n",
    "    ASSERT(obj_block->is_free_, \"Expected free ObjectBlock at index:\" +\n",
    "                                    std::to_string(next_free_index_));\n",
    "\n",
    "    T* ret = &(obj_block->object_);  // 获取槽位中 T 的原生指针\n",
    "    ret = new (ret) T(args...);      // placement new：在已有内存上构造 T\n",
    "    obj_block->is_free_ = false;     // 标记槽位为占用状态\n",
    "\n",
    "    updateNextFreeIndex();  // 更新下一个空闲槽位索引\n",
    "    return ret;\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "接下来看看`updateNextFreeIndex()`方法。这里有两点需要注意\n",
    "- 第一，我们有一个处理索引绕回末尾情况的分支。虽然这里添加了一个`if`条件，但通过`UNLIKELY()`说明以及我们硬件分支预测器通常会预测该分支不会被执行，这应该不会对性能产生太大影响。当然，如果真的想的话，我们可以把循环分成两个循环，去掉这个`if`条件——也就是说，第一个循环一直到`next_free_index_ == store_.size()`，第二个循环从0开始。\n",
    "- 第二，我们添加了一个检查，以检测并在内存池完全满了的情况下报错。在实际中显然有更好的处理方式，不会导致程序出错，但为了简洁起见，并且在本书的讨论范围内，我们现在就只是在这种情况发生时报错："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "/**\n",
    " * @brief 寻找下一个可用槽位。\n",
    " *\n",
    " *  - 从当前 `next_free_index_` 开始向前扫描。\n",
    " *  - 若到达末尾，则回绕至 0（循环队列思路）。\n",
    " *  - 如果完整扫描一圈仍然没有找到空闲槽位，则触发断言，提示池已耗尽。\n",
    " *\n",
    " * @note 常见路径是“快速命中空闲槽位”，因此配合 `LIKELY/UNLIKELY`\n",
    " * 提示编译器进行分支预测优化。\n",
    " */\n",
    "auto updateNextFreeIndex() noexcept {\n",
    "    const auto initial_free_index = next_free_index_;\n",
    "    while (!store_[next_free_index_].is_free_) {\n",
    "        ++next_free_index_;\n",
    "        if (UNLIKELY(next_free_index_ ==\n",
    "                     store_.size())) {  // 回绕到起始位置（该分支应当很少触发）\n",
    "            next_free_index_ = 0;\n",
    "        }\n",
    "        if (UNLIKELY(initial_free_index == next_free_index_)) {\n",
    "            // 断言：若重新回到起始索引，说明没有可用槽位，内存池已用尽。\n",
    "            ASSERT(initial_free_index != next_free_index_,\n",
    "                   \"Memory Pool out of space.\");\n",
    "        }\n",
    "    }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "下一部分将处理内存释放，即将`T`类型的对象返回内存池，以便回收为空闲状态。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 处理内存释放\n",
    "\n",
    "释放内存就是在内部的`store_`中找到与正在释放的`T`对象对应的正确`ObjectBlock`，并将该块的`is_free_`标记设置为`true`。这里，我们使用`reinterpret_cast`将`T*`转换为`ObjectBlock*`，这是可行的，因为对象`T`是`ObjectBlock`中的第一个成员。这就解释了我们在“初始化内存池”部分的构造函数中添加的断言。我们在这里也添加了一个断言，以确保用户试图释放的元素属于这个内存池。同样，对于这种错误情况可以有更优雅的处理方式，但为了简洁并保持在本书的讨论范围内，我们把这个留给你去探索："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "/**\n",
    " * @brief 回收指定指针指向的对象槽位。\n",
    " *\n",
    " * @param elem 指向池内对象的指针。\n",
    " * @note 仅将槽位标记为空闲，不调用对象析构函数；使用者需要自行管理资源释放。\n",
    " */\n",
    "auto deallocate(const T* elem) noexcept {\n",
    "    // 将 T* 强制解释为 ObjectBlock*，然后通过指针差值计算索引。\n",
    "    const auto elem_index = (reinterpret_cast<const ObjectBlock*>(elem) - &store_[0]);\n",
    "\n",
    "    // 断言：检查待回收指针是否确实来自当前内存池。\n",
    "    ASSERT(elem_index >= 0 && static_cast<size_t>(elem_index) < store_.size(),\n",
    "           \"Element being deallocated does not belong to this Memory pool.\");\n",
    "    // 断言：确保该槽位当前处于“已占用”状态，防止重复回收导致逻辑错误。\n",
    "    ASSERT(\n",
    "        !store_[elem_index].is_free_,\n",
    "        \"Expected in-use ObjectBlock at index:\" + std::to_string(elem_index));\n",
    "\n",
    "    store_[elem_index].is_free_ = true;  // 仅重置空闲标记\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "以上就是我们内存池的设计与实现。让我们来看一个简单的例子。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 结合示例使用内存池"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "让我们来看一个简单且易于理解的示例，展示刚刚创建的内存池的使用方法。这段代码在[`Chapter4/mem_pool_example.cpp`](https://github.com/PacktPublishing/Building-Low-Latency-Applications-with-CPP/blob/main/Chapter4/mem_pool_example.cpp)文件中，如前文所述，可以使用CMake文件进行构建。它创建了一个基本类型`double`的内存池和另一个自定义`MyStruct`类型的内存池。然后，从这个内存池分配和释放一些元素，并打印出它们的值和内存地址："
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- [mem_pool_example.cpp](./mem_pool_example/mem_pool_example.cpp)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "使用以下命令运行这个示例，应该会产生类似如下的输出："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "! cd ../../ && rm -rf build && cmake -S . -B build && cmake --build build --target mem_pool_example --config Release "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "! cd ../../build/Part_1/04/mem_pool_example && ./mem_pool_example"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "在下一节中，我们将构建一个非常类似的组件——无锁队列（lock-free queues）。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "对于无锁队列，我们同样可以选择在栈上或堆上分配存储。这里，我们再次选择`std::vector`并在堆上分配内存。此外，我们创建两个`std::atomic`变量，一个名为`next_write_index_`，用于跟踪下一次向队列写入数据的索引位置；第二个变量名为`next_read_index_`，用于跟踪队列中下一个未读元素所在的索引位置。由于我们假设只有一个线程写入队列，一个线程读取队列，所以实现相对简单。本节讨论的源代码可以在本书GitHub仓库的[`Chapter4/lf_queue.h`](https://github.com/PacktPublishing/Building-Low-Latency-Applications-with-CPP/blob/main/Chapter4/lf_queue.h)源文件中找到。\n",
    "\n",
    "关于`std::atomic`简单说明一下，它是现代C++的一种结构，允许进行线程安全的操作。它让我们在不使用锁或互斥锁的情况下，对共享变量进行读取、更新和写入操作，并且在操作过程中保持操作顺序。关于`std::atomic`和内存序的详细讨论超出了本书的范围，你可以在我们的另一本书《开发高频交易系统》中找到参考资料。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "首先，我们在以下代码片段中定义这个类的数据成员："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "std::vector<T> store_; ///< 环形缓冲区的实际存储容器。\n",
    "\n",
    "std::atomic<size_t> next_write_index_ = {0}; ///< 下一次写入的位置索引。\n",
    "std::atomic<size_t> next_read_index_ = {0};  ///< 下一次读取的位置索引。\n",
    "\n",
    "std::atomic<size_t> num_elements_ = {0};     ///< 当前已占用的元素数量。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "这个类包含一个`std::vector`对象`store_`，其类型为`T`模板对象，它是实际的数据队列。`std::atomic<size_t>`类型的`next_write_index_`变量跟踪向量中下次写入元素的索引位置。类似地，`std::atomic<size_t>`类型的`next_read_index_`变量跟踪向量中下次读取或消费元素的索引位置。由于读取和写入操作是由不同线程执行的，所以这些变量需要是`std::atomic<>`类型。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 初始化无锁队列\n",
    "\n",
    "无锁队列的构造函数与我们之前看到的内存池的构造函数非常相似。我们在构造函数中动态分配整个向量的内存。我们可以扩展这个设计，使无锁队列在运行时能够调整大小，但目前我们将使用固定大小的队列："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "/**\n",
    " * @brief 构造指定容量的环形队列。\n",
    " *\n",
    " * @param num_elems 预先分配的存储槽位数量。\n",
    " * @note 使用 std::vector<T> 进行底层存储，保证元素空间连续并便于缓存友好访问。\n",
    " */\n",
    "explicit LFQueue(std::size_t num_elems)\n",
    "    : store_(num_elems, T()) /* 预分配 vector 容量，并以 T 的默认值初始化。 */ {\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "关于默认构造函数、复制构造函数、移动构造函数以及赋值运算符，我们有类似的样板代码。出于之前讨论的原因，这些函数被删除："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// 禁用默认构造、拷贝/移动构造与赋值，防止不受控的资源复制。\n",
    "LFQueue() = delete;\n",
    "LFQueue(const LFQueue &) = delete;\n",
    "LFQueue(const LFQueue &&) = delete;\n",
    "LFQueue &operator=(const LFQueue &) = delete;\n",
    "LFQueue &operator=(const LFQueue &&) = delete;"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "接下来，我们将查看向队列添加新元素的代码。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 向队列添加元素\n",
    "\n",
    "向队列添加新元素的代码分两部分实现：\n",
    "- 第一部分`getNextToWriteTo()`返回一个指针，指向要写入新数据的下一个元素；\n",
    "- 第二部分`updateWriteIndex()`在元素写入提供的槽位后，递增写入索引`next_write_index_`。\n",
    "\n",
    "我们这样设计，是为了取代单一的`write()`函数，为用户提供指向元素的指针。如果对象很大，就无需更新或覆盖整个对象。此外，这种设计也更容易处理竞态条件："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "/**\n",
    " * @brief 获取当前可写入的位置指针。\n",
    " *\n",
    " * @return 指向可写槽位的原生指针。\n",
    " * @note 调用者负责在写入数据后显式调用 updateWriteIndex() 推进写指针。\n",
    " */\n",
    "auto getNextToWriteTo() noexcept { return &store_[next_write_index_]; }\n",
    "\n",
    "/**\n",
    " * @brief 将写索引推进到下一个槽位，并更新元素计数。\n",
    " *\n",
    " * @note\n",
    " *  - 采用取模操作实现索引循环。\n",
    " *  - num_elements_++ 使用原子递增，保证多线程下计数语义正确。\n",
    " */\n",
    "auto updateWriteIndex() noexcept {\n",
    "    next_write_index_ = (next_write_index_ + 1) % store_.size();\n",
    "    num_elements_++;\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "在下一节中，我们将使用非常相似的设计从队列消费元素。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 从队列消费元素\n",
    "\n",
    "从队列消费元素的操作与向队列添加元素的操作相反。就像我们将`write()`函数拆分为两部分的设计一样，从队列消费元素也分为两部分。我们有一个`getNextToRead()`方法，它返回一个指向要消费的下一个元素的指针，但不会更新读取索引。如果没有元素可消费，这个方法将返回`nullptr`。第二部分`updateReadIndex()`在元素被消费后更新读取索引：\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "/**\n",
    " * @brief 获取当前可读元素的指针。\n",
    " *\n",
    " * @return 若队列非空，返回指向当前读索引处元素的指针；否则返回 nullptr。\n",
    " * @note 调用者需要在消费元素后调用 updateReadIndex() 推进读指针。\n",
    " */\n",
    "auto getNextToRead() const noexcept -> const T* {\n",
    "    return (size() ? &store_[next_read_index_] : nullptr);\n",
    "}\n",
    "\n",
    "/**\n",
    " * @brief 将读索引推进到下一个槽位，并递减元素计数。\n",
    " *\n",
    " * @note\n",
    " *  - 读索引同样采用取模实现循环。\n",
    " *  - 通过断言确保在合法的非空状态下执行读取，便于调试阶段捕获错误。\n",
    " */\n",
    "auto updateReadIndex() noexcept {\n",
    "    next_read_index_ = (next_read_index_ + 1) % store_.size();\n",
    "    ASSERT(num_elements_ != 0,\n",
    "           \"Read an invalid element in:\" + std::to_string(pthread_self()));\n",
    "    num_elements_--;\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "我们还定义了另一个简单的方法来返回队列中的元素数量："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "/**\n",
    " * @brief 当前队列中元素数量。\n",
    " *\n",
    " * @return 以原子方式读取的元素计数。\n",
    " */\n",
    "auto size() const noexcept {\n",
    "  return num_elements_.load();\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "至此，我们完成了单生产者单消费者场景下无锁队列的设计与实现。在下一小节中，我们来看一个使用这个组件的示例。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 使用无锁队列"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "如何使用无锁数据队列的示例可以在[`Chapter4/lf_queue_example.cpp`](https://github.com/PacktPublishing/Building-Low-Latency-Applications-with-CPP/blob/main/Chapter4/lf_queue_example.cpp)文件中找到，并且可以按照前面提到的方式进行构建。这个示例创建了一个消费者线程，并为其提供一个无锁队列实例。然后生产者生成并向该队列添加一些元素，消费者线程检查队列并消费队列中的元素，直到队列为空。生产者和消费者这两个执行线程在生成和消费元素之间都会短暂等待：\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- [lf_queue_example.cpp](./lf_queue_example/lf_queue_example.cpp)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "! cd ../../ && rm -rf build && cmake -S . -B build && cmake --build build --target lf_queue_example --config Release "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "! cd ../../build/Part_1/04/lf_queue_example && ./lf_queue_example"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "接下来，我们将使用刚刚构建的一些组件（线程和无锁队列）来构建一个低延迟日志框架。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 构建低延迟日志框架\n",
    "\n",
    "现在，我们将使用上几节构建的一些组件来构建一个低延迟日志框架。日志记录是任何应用程序的重要组成部分，无论是记录一般的应用程序行为、警告、错误，还是性能统计信息。然而，许多重要的日志输出实际上来自关键路径上对性能要求很高的组件。\n",
    "\n",
    "一种简单的日志记录方法是输出到屏幕上，而稍微好一点的方法是将日志保存到一个或多个日志文件中。但这里存在一些问题——磁盘I/O极其缓慢且不可预测，字符串操作和格式化本身也很慢。由于这些原因，在对性能要求很高的线程上执行这些操作是个糟糕的主意，所以在本节中，我们将构建一个解决方案，在保留按需输出日志能力的同时，减轻这些缺点。\n",
    "\n",
    "在深入研究日志记录器类之前，我们将定义一些实用方法，用于获取当前系统时间并将其转换为字符串以便记录日志。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 设计时间相关的实用方法\n",
    "\n",
    "我们将定义一个简单的实用函数来获取当前系统时间，并定义一些常量，以便更轻松地进行不同单位之间的转换。时间实用函数的代码可以在本书GitHub仓库的[`Chapter4/time_utils.h`](https://github.com/PacktPublishing/Building-Low-Latency-Applications-with-CPP/blob/main/Chapter4/time_utils.h)文件中找到："
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- [time_utils.h](./libcommon/time_utils.h)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "现在，让我们从下一节开始设计日志记录器类本身。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 设计低延迟日志记录器\n",
    "\n",
    "为了构建这个低延迟日志框架，我们将创建一个后台日志记录线程，其唯一任务是将日志行写入磁盘上的日志文件。这样做的目的是将缓慢的磁盘I/O操作以及字符串格式化操作从对性能要求很高的主线程转移到这个后台线程。需要理解的是，将日志写入磁盘并不需要即时完成——也就是说，大多数系统能够容忍事件发生与将该事件相关的信息记录到磁盘之间存在一定延迟。我们将使用本章第一节创建的多线程函数来创建这个日志记录线程，并为其分配写入日志文件的任务。\n",
    "\n",
    "为了将需要记录的日志数据从对性能要求很高的主线程发布到这个日志记录线程，我们将使用上一节创建的无锁数据队列。日志记录器的工作方式是，对性能敏感的线程不会直接将信息写入磁盘，而是简单地将信息推送到这个无锁队列中。正如我们之前讨论的，日志记录线程将从这个队列的另一端获取数据并写入磁盘。该组件的源代码可以在本书GitHub仓库`Chapter4`目录下的[`logging.h`](https://github.com/PacktPublishing/Building-Low-Latency-Applications-with-CPP/blob/main/Chapter4/logging.h)文件中找到。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- [logging.h](./libcommon/logging.h)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 定义一些日志记录器结构\n",
    "\n",
    "在开始设计日志记录器本身之前，我们首先定义将通过无锁队列从对性能敏感的线程传输到日志记录线程的基本信息块。在这个设计中，我们简单地创建一个能够保存我们要记录的不同类型数据的结构。首先，让我们定义一个枚举，用于指定它所指向的结构的值的类型；我们将这个枚举称为`LogType`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "enum class LogType : int8_t {\n",
    "  CHAR = 0,\n",
    "  INTEGER = 1,\n",
    "  LONG_INTEGER = 2,\n",
    "  LONG_LONG_INTEGER = 3,\n",
    "  UNSIGNED_INTEGER = 4,\n",
    "  UNSIGNED_LONG_INTEGER = 5,\n",
    "  UNSIGNED_LONG_LONG_INTEGER = 6,\n",
    "  FLOAT = 7,\n",
    "  DOUBLE = 8\n",
    "};"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "现在，我们可以定义`LogElement`结构，它将保存下一个要推送到队列中的值，并最终由日志记录线程将日志写入文件。这个结构包含一个`LogType`类型的成员，用于指定它保存的值的类型。这个结构中的另一个成员是不同可能基本类型的联合体。在现代C++中，这里本可以使用`std::variant`，因为它是一个类型安全的联合体，并且内置了判别器（`LogType type_`，用于指定联合体包含的内容）。然而，`std::variant`的运行时性能较差，因此我们在这里选择使用旧式联合体："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "struct LogElement {\n",
    "  LogType type_ = LogType::CHAR;\n",
    "  union {\n",
    "    char c;\n",
    "    int i;\n",
    "    long l;\n",
    "    long long ll;\n",
    "    unsigned u;\n",
    "    unsigned long ul;\n",
    "    unsigned long long ull;\n",
    "    float f;\n",
    "    double d;\n",
    "  } u_;\n",
    "};"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "定义好`LogElement`结构后，让我们继续定义日志记录器类中的数据。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 初始化日志记录器数据结构"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "我们的日志记录器将包含几个重要的对象。首先，`std::ofstream`文件对象是用于写入数据的日志文件。其次，`LFQueue<LogElement>`对象是用于将数据从主线程传输到日志记录线程的无锁队列。接下来，`std::atomic<bool>`用于在需要时停止日志记录线程的处理，还有一个`std::thread`对象，它就是日志记录线程。最后，`std::string`是文件名，我们仅将其用于提供信息："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "private:\n",
    "  const std::string file_name_;       ///< 日志文件名。\n",
    "  std::ofstream file_;                ///< 输出文件流句柄。\n",
    "  LFQueue<LogElement> queue_;         ///< 日志缓冲队列。\n",
    "  std::atomic<bool> running_ = {true};///< 后台线程运行标志。\n",
    "  std::thread *logger_thread_ = nullptr; ///< 后台写线程指针。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "现在，让我们开始构建日志记录器、日志记录器队列和日志记录器线程。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 创建日志记录器并启动日志记录器线程"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "在日志记录器的构造函数中，我们将使用合适的大小初始化日志记录器队列，保存`file_name_`用于提供信息，打开输出日志文件对象，并创建和启动日志记录器线程。请注意，如果无法打开输出日志文件或无法创建和启动日志记录器线程，我们将在此处退出。正如我们之前提到的，显然有更宽容、更优雅的方式来处理这些失败情况，但在本书中我们不会探讨这些内容。这里需要注意的是，我们在`createAndStartThread()`中将`core_id`参数设置为 -1，目前暂不设置线程的亲和性。在本书后面，一旦我们了解了整个生态系统的设计，我们将重新讨论如何为每个线程分配CPU核心的设计，并对其进行性能调优："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "  /**\n",
    "   * @brief 构造 Logger 并立即启动后台线程。\n",
    "   *\n",
    "   * @param file_name 日志文件名。\n",
    "   *\n",
    "   * @throws std::runtime_error（通过 ASSERT）若文件打开失败或线程创建失败。\n",
    "   */\n",
    "  explicit Logger(const std::string &file_name)\n",
    "      : file_name_(file_name), queue_(LOG_QUEUE_SIZE) {\n",
    "    file_.open(file_name);\n",
    "    ASSERT(file_.is_open(), \"Could not open log file:\" + file_name);\n",
    "    logger_thread_ = createAndStartThread(-1, \"Common/Logger \" + file_name_,\n",
    "                                          [this]() { flushQueue(); });\n",
    "    ASSERT(logger_thread_ != nullptr, \"Failed to start Logger thread.\");\n",
    "  }\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "我们传递了一个名为`flushQueue()`的方法，这个日志记录线程将运行该方法。顾名思义，并且与我们之前讨论的内容一致，这个线程将清空日志数据队列，并将数据写入文件；接下来我们将了解这个方法。`flushQueue()`的实现很简单。如果原子布尔变量`running_`为`true`，它将在一个循环中运行，执行以下步骤：它获取推送到无锁队列`queue_`中的任何新元素，并将它们写入我们创建的`file_`对象。它根据类型解包队列中的`LogElement`对象，并将联合体中正确的成员写入文件。当无锁队列空时，线程会睡眠一毫秒，然后再次检查是否有新元素需要写入磁盘："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "  /**\n",
    "   * @brief 后台线程入口：持续从队列读取日志元素并写入文件。\n",
    "   *\n",
    "   * @note\n",
    "   *  - 当 running_ 由外部置为 false 时循环结束。\n",
    "   *  - 每轮消费完数据后主动 flush，确保数据及时落盘。\n",
    "   */\n",
    "  auto flushQueue() noexcept {\n",
    "    while (running_) {\n",
    "      for (auto next = queue_.getNextToRead(); queue_.size() && next;\n",
    "           next = queue_.getNextToRead()) {\n",
    "        switch (next->type_) {\n",
    "        case LogType::CHAR:\n",
    "          file_ << next->u_.c;\n",
    "          break;\n",
    "        case LogType::INTEGER:\n",
    "          file_ << next->u_.i;\n",
    "          break;\n",
    "        case LogType::LONG_INTEGER:\n",
    "          file_ << next->u_.l;\n",
    "          break;\n",
    "        case LogType::LONG_LONG_INTEGER:\n",
    "          file_ << next->u_.ll;\n",
    "          break;\n",
    "        case LogType::UNSIGNED_INTEGER:\n",
    "          file_ << next->u_.u;\n",
    "          break;\n",
    "        case LogType::UNSIGNED_LONG_INTEGER:\n",
    "          file_ << next->u_.ul;\n",
    "          break;\n",
    "        case LogType::UNSIGNED_LONG_LONG_INTEGER:\n",
    "          file_ << next->u_.ull;\n",
    "          break;\n",
    "        case LogType::FLOAT:\n",
    "          file_ << next->u_.f;\n",
    "          break;\n",
    "        case LogType::DOUBLE:\n",
    "          file_ << next->u_.d;\n",
    "          break;\n",
    "        }\n",
    "        queue_.updateReadIndex();\n",
    "      }\n",
    "      file_.flush();\n",
    "\n",
    "      using namespace std::literals::chrono_literals;\n",
    "      std::this_thread::sleep_for(10ms);\n",
    "    }\n",
    "  }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "日志记录器类的析构函数很重要，让我们看看它需要执行哪些清理任务。首先，析构函数等待无锁队列被日志记录线程处理完，即等待队列变空。一旦队列空了，它将`running_`标志设置为`false`，以便日志记录线程可以完成执行。为了等待日志记录线程完成执行，即从`flushQueue()`方法返回，它调用日志记录线程的`std::thread::join()`方法。最后，它关闭`file_`对象，这会将任何缓冲的数据写入磁盘，然后就完成了："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "  /**\n",
    "   * @brief 析构时先等待队列清空，再停止后台线程并关闭文件。\n",
    "   *\n",
    "   * @note\n",
    "   *  - 通过轮询等待队列 size() 为 0，避免丢失尚未写出的日志。\n",
    "   *  - join() 保证线程安全退出。\n",
    "   */\n",
    "  ~Logger() {\n",
    "    std::string time_str;\n",
    "    std::cerr << Common::getCurrentTimeStr(&time_str)\n",
    "              << \" Flushing and closing Logger for \" << file_name_ << std::endl;\n",
    "\n",
    "    while (queue_.size()) {\n",
    "      using namespace std::literals::chrono_literals;\n",
    "      std::this_thread::sleep_for(1s);\n",
    "    }\n",
    "    running_ = false;\n",
    "    logger_thread_->join();\n",
    "\n",
    "    file_.close();\n",
    "    std::cerr << Common::getCurrentTimeStr(&time_str) << \" Logger for \"\n",
    "              << file_name_ << \" exiting.\" << std::endl;\n",
    "  }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "最后，我们添加之前多次讨论过的关于构造函数和赋值运算符的常规样板代码："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "  // 禁止默认构造以及拷贝/移动语义，避免多实例共享资源。\n",
    "  Logger() = delete;\n",
    "  Logger(const Logger &) = delete;\n",
    "  Logger(const Logger &&) = delete;\n",
    "  Logger &operator=(const Logger &) = delete;\n",
    "  Logger &operator=(const Logger &&) = delete;"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "在本节中，我们了解了从队列获取数据并将其写入磁盘的组件部分。在下一节中，我们将了解在日志记录过程中，对性能敏感的线程是如何将数据添加到无锁队列中的。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 将数据推送到日志记录器队列"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "为了将数据推送到日志记录器队列，我们将定义几个重载的`pushValue()`方法来处理不同类型的参数。每个方法的作用都是将值逐个推送到队列中。这里值得注意的是，对于我们即将讨论的内容，有更高效的实现方式；然而，它们会涉及额外的复杂性，为了简洁起见，并限制本书的讨论范围，我们省略了这些内容。在讨论时，我们会指出潜在的改进之处。\n",
    "\n",
    "首先，我们创建一个`pushValue()`的变体来推送`LogElement`类型的对象，它将被我们稍后定义的其他`pushValue()`函数调用。它基本上是将数据写入无锁队列的下一个位置，并增加写入索引："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "  /**\n",
    "   * @brief 将单个日志元素写入环形队列。\n",
    "   *\n",
    "  * @param log_element 预先封装好的日志数据。\n",
    "   */\n",
    "  auto pushValue(const LogElement &log_element) noexcept {\n",
    "    *(queue_.getNextToWriteTo()) = log_element;\n",
    "    queue_.updateWriteIndex();\n",
    "  }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "下一个简单的`pushValue()`变体用于单个`char`值，它基本上只是创建一个`LogElement`类型的对象，调用我们刚刚讨论的`pushValue()`方法，并传递这个`LogElement`对象："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "  /** @name 原生类型写入接口 */\n",
    "  ///@{\n",
    "  auto pushValue(const char value) noexcept {\n",
    "    pushValue(LogElement{LogType::CHAR, {.c = value}});\n",
    "  }\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "现在，我们创建一个用于`const char*`（即字符集合）的`pushValue()`变体。这个实现会逐个遍历字符，并调用我们之前实现的`pushValue()`。这是一个有潜在改进空间的地方，我们可以使用单个`memcpy()`来复制数组中的所有字符，而不是逐个遍历它们。在队列末尾索引回绕的情况下，我们需要处理一些边界情况，但这部分内容留给你进一步探索："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "  /**\n",
    "   * @brief 写入字符串（C 风格）。\n",
    "   *\n",
    "   * @param value 以 '\\0' 结尾的 C 字符串，逐字符写入队列。\n",
    "   */\n",
    "  auto pushValue(const char *value) noexcept {\n",
    "    while (*value) {\n",
    "      pushValue(*value);\n",
    "      ++value;\n",
    "    }\n",
    "  }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "接下来，我们创建一个用于`const std::string&`的`pushValue()`变体，这个变体非常直接，它使用了我们之前创建的`pushValue()`："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "  /**\n",
    "   * @brief 写入 std::string。\n",
    "   *\n",
    "   * @param value 字符串对象，内部复用 C 字符串接口。\n",
    "   */\n",
    "  auto pushValue(const std::string &value) noexcept {\n",
    "    pushValue(value.c_str());\n",
    "  }\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "最后，我们需要为不同的基本类型添加`pushValue()`的变体。它们与我们为单个`char`值构建的变体非常相似，如下所示："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "  auto pushValue(const int value) noexcept {\n",
    "    pushValue(LogElement{LogType::INTEGER, {.i = value}});\n",
    "  }\n",
    "\n",
    "  // ...\n",
    "\n",
    "  auto pushValue(const double value) noexcept {\n",
    "    pushValue(LogElement{LogType::DOUBLE, {.d = value}});\n",
    "  }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "此时，我们实现了两个目标——将磁盘输出操作转移到后台日志记录线程，并将将基本类型值格式化为字符串格式的任务转移到后台线程。接下来，我们将添加对性能敏感的线程使用的功能，以便通过我们刚刚构建的`pushValue()`方法将数据推送到无锁队列。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 添加一个实用的通用日志函数"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "我们将定义一个`log()`方法，它与`printf()`函数非常相似，但稍微简单一些。说它简单是因为格式说明符只有一个`%`字符，用于替代所有不同的基本类型。这个方法使用可变参数模板 (variadic template arguments)来支持任意数量和类型的参数。它会查找`%`字符，然后用下一个值替换它，并调用上一节定义的某个重载的`pushValue()`方法。之后，它会递归调用自身，不过这一次，值指向模板参数包中的第一个参数：\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "  /**\n",
    "   * @brief 简易格式化输出：顺序替换字符串中的占位符。\n",
    "   *\n",
    "   * @tparam T 第一个参数类型。\n",
    "   * @tparam A 剩余参数类型列表。\n",
    "   * @param s   格式字符串，使用 % 占位，%% 表示转义的百分号。\n",
    "   * @param value 第一个替换值。\n",
    "   * @param args  剩余替换值。\n",
    "   *\n",
    "   * @note 若占位符与参数个数不匹配，将调用 FATAL 终止程序。\n",
    "   */\n",
    "  template <typename T, typename... A>\n",
    "  auto log(const char *s, const T &value, A... args) noexcept {\n",
    "    while (*s) {\n",
    "      if (*s == '%') {\n",
    "        if (UNLIKELY(*(s + 1) == '%')) {\n",
    "          ++s;\n",
    "        } else {\n",
    "          pushValue(value);\n",
    "          log(s + 1, args...);\n",
    "          return;\n",
    "        }\n",
    "      }\n",
    "      pushValue(*s++);\n",
    "    }\n",
    "    FATAL(\"extra arguments provided to log()\");\n",
    "  }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "这个方法可以像下面这个例子这样调用："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "int int_val = 10;\n",
    "std::string str_val = \"hello\";\n",
    "double dbl_val = 10.10;\n",
    "log(\"Integer:% String:% Double:%\", int_val, str_val, dbl_val);"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "我们这里构建的`log()`方法无法处理没有传递参数的情况。因此，我们需要一个额外的重载`log()`方法来处理只传递一个`const char *`的情况。我们在这里添加了一个额外的检查，以确保没有给这个方法或前面提到的`log()`方法传递多余的参数："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "  /**\n",
    "   * @brief 无参数版本，确保格式串中不存在未替换的占位符。\n",
    "   *\n",
    "   * @param s 格式字符串。\n",
    "   */\n",
    "  auto log(const char *s) noexcept {\n",
    "    while (*s) {\n",
    "      if (*s == '%') {\n",
    "        if (UNLIKELY(*(s + 1) == '%')) {\n",
    "          ++s;\n",
    "        } else {\n",
    "          FATAL(\"missing arguments to log()\");\n",
    "        }\n",
    "      }\n",
    "      pushValue(*s++);\n",
    "    }\n",
    "  }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "至此，我们完成了低延迟日志框架的设计与实现。利用多线程例程和无锁队列，我们创建了一个框架，让对性能要求高的线程将字符串格式化和磁盘文件写入任务转移到后台日志线程。现在，让我们看一个如何创建、配置和使用刚创建的日志记录器的示例。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 通过示例学习如何使用日志记录器"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "我们将展示一个基本示例，创建一个`Logger`对象并配置它将日志写入`logging_example.log`文件。然后，它通过日志记录器将几种不同的数据类型记录到文件中。示例代码可以在本书GitHub仓库的[`Chapter4/logging_example.cpp`](https://github.com/PacktPublishing/Building-Low-Latency-Applications-with-CPP/blob/main/Chapter4/logging_example.cpp)文件中找到："
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- [logging_example.cpp](./logging_example/logging_example.cpp)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "运行该示例后，通过输出当前目录下`logging_example.log`文件的内容，可查看输出结果，如下所示："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "! cd ../../ && rm -rf build && cmake -S . -B build && cmake --build build --target logging_example --config Release "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Set core affinity for Common/Logger logging_example.log 135248289457728 to -1\n",
      "Fri Oct 10 05:14:17 2025Fri Oct 10 05:14:18 2025"
     ]
    }
   ],
   "source": [
    "! cd ../../build/Part_1/04/logging_example && ./logging_example"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Logging a char:d an int:3 and an unsigned:65\n",
      "Logging a float:3.4 and a double:34.56\n",
      "Logging a C-string:'test C-string'\n",
      "Logging a string:'test string'\n"
     ]
    }
   ],
   "source": [
    "! cd ../../build/Part_1/04/logging_example && cat logging_example.log"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "在这个框架中，调用`log()`的唯一开销是遍历字符串中的字符并将字符和值推送到无锁队列的开销。现在，我们将讨论转移到网络编程和套接字的使用上，后续我们将使用它们来实现不同进程之间的通信。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 使用套接字进行C++网络编程"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "在本节中，我们将构建最后一个基础组件——一个使用Unix套接字进行网络编程的框架。我们将使用这个框架构建一个监听传入TCP连接的服务器，以及一个能够与该服务器建立TCP连接的客户端。我们还将使用这个框架来发布UDP流量并从多播流量流中接收数据。请注意，为了限制讨论范围，我们只讨论没有内核旁路功能的Unix套接字。使用内核旁路并利用支持它的网络接口卡（Network Interface Card，NIC）提供的内核旁路API超出了本书的范围。另外，我们假设你对网络套接字有一些基本的了解或经验，最好有使用C++进行网络套接字编程的经验。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 构建基本的套接字API\n",
    "\n",
    "我们的目标是创建一种机制，用于创建网络套接字并用正确的参数对其进行初始化。这个方法将用于创建监听、接收和发送套接字，以便通过UDP和TCP协议进行通信。在深入研究创建套接字的例程之前，让我们先定义一些在最终方法中会用到的实用方法。基本套接字API的所有代码都在本书GitHub仓库的[`Chapter4/socket_utils.h`](https://github.com/PacktPublishing/Building-Low-Latency-Applications-with-CPP/blob/main/Chapter4/socket_utils.h)文件中。在研究功能实现之前，我们先展示`Chapter4/socket_utils.h`头文件，其中包含了所有要实现的包含文件和函数签名："
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- [socket_utils.h](./libcommon/socket_utils.h)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "现在，让我们从下一节开始实现这些方法。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "我们需要构建的第一个实用方法是将以字符串形式表示的网络接口转换为底层套接字例程可以使用的形式。我们将这个方法称为`getIfaceIP()`，在指定监听、连接或发送数据的网络接口时会用到它。我们使用`getifaddrs()`方法获取所有接口的信息，它返回一个包含这些信息的链表结构`ifaddrs`。最后，使用`getnameinfo()`信息获取最终要与其他方法一起使用的名称："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "  /**\n",
    "   * @brief 根据网卡名获取其 IPv4 地址。\n",
    "   *\n",
    "   * @param iface 网卡名（如 \"eth0\"）。\n",
    "   * @return 对应的数值型 IP 字符串；若未找到或出错返回空串。\n",
    "   *\n",
    "   * @note\n",
    "   *  - 此函数与日志模块无直接耦合，但在 createSocket 中会配合 Logger 记录结果。\n",
    "   */\n",
    "  inline auto getIfaceIP(const std::string &iface) -> std::string {\n",
    "    char buf[NI_MAXHOST] = {'\\0'};\n",
    "    ifaddrs *ifaddr = nullptr;\n",
    "\n",
    "    if (getifaddrs(&ifaddr) != -1) {\n",
    "      for (ifaddrs *ifa = ifaddr; ifa; ifa = ifa->ifa_next) {\n",
    "        if (ifa->ifa_addr && ifa->ifa_addr->sa_family == AF_INET && iface == ifa->ifa_name) {\n",
    "          getnameinfo(ifa->ifa_addr, sizeof(sockaddr_in), buf, sizeof(buf), NULL, 0, NI_NUMERICHOST);\n",
    "          break;\n",
    "        }\n",
    "      }\n",
    "      freeifaddrs(ifaddr);\n",
    "    }\n",
    "\n",
    "    return buf;\n",
    "  }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "例如，在我的系统上有以下网络接口，结果如下："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "lo: flags=73<UP,LOOPBACK,RUNNING>  mtu 65536\n",
    "inet 127.0.0.1  netmask 255.0.0.0\n",
    "wlp4s0: flags=4163<UP,BROADCAST,RUNNING,MULTICAST>  mtu 1500\n",
    "inet 192.168.10.104  netmask 255.255.255.0  broadcast 192.168.10.255"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`getIfaceIP(\"lo\")`返回`127.0.0.1`，`getIfaceIP(\"wlp4s0\")`返回`192.168.10.104`。\n",
    "\n",
    "接下来，我们将介绍下一个重要的实用函数，它会影响需要网络套接字的应用程序的性能。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 设置套接字为非阻塞模式"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "我们要构建的下一个实用函数是将套接字设置为非阻塞模式。阻塞套接字是指对其进行读取调用时，会无限期阻塞，直到有数据可用。对于极低延迟的应用程序来说，这通常不是一个好的设计，原因有很多。主要原因之一是阻塞套接字是通过在用户空间和内核空间之间进行切换来实现的，这效率非常低。当套接字需要被唤醒或解除阻塞时，需要从内核空间到用户空间产生一个中断、一个中断处理程序等来处理该事件。此外，被阻塞的对性能要求高的线程会产生上下文切换成本，正如前面所讨论的，这对性能是有害的。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "下面的`setNonBlocking()`方法使用`fcntl()`例程和`F_GETFL`首先检查套接字文件描述符，查看它是否已经是非阻塞的。如果不是，那么再次使用`fcntl()`例程，但这次使用`F_SETFL`来添加非阻塞标志位，该标志位会设置在文件描述符上。如果套接字文件描述符已经是非阻塞的，或者该方法成功将其设置为非阻塞，则返回`true`："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "  /**\n",
    "   * @brief 将 socket 设置为非阻塞模式。\n",
    "   *\n",
    "   * @param fd 有效的 socket 文件描述符。\n",
    "   * @return 设置成功返回 true，失败返回 false。\n",
    "   *\n",
    "   * @note\n",
    "   *  - 若需要记录失败原因，可调用 Logger 将 errno 信息写入日志文件。\n",
    "   */\n",
    "  inline auto setNonBlocking(int fd) -> bool {\n",
    "    const auto flags = fcntl(fd, F_GETFL, 0);\n",
    "    if (flags & O_NONBLOCK)\n",
    "      return true;\n",
    "    return (fcntl(fd, F_SETFL, flags | O_NONBLOCK) != -1);\n",
    "  }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "接下来，我们将通过禁用Nagle算法为TCP套接字实现另一项重要的优化。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 禁用Nagle算法"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "在此不过多深入细节，Nagle算法用于优化TCP套接字的缓冲机制，并避免TCP套接字在保证可靠性时产生的额外开销。它通过延迟发送部分数据包，而非立即发送来实现这一目的。对于许多应用程序而言，这是一个不错的特性，但对于低延迟应用程序来说，消除发送数据包时的延迟至关重要。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "幸运的是，禁用Nagle算法很简单，只需使用`setsockopt()`函数设置套接字选项`TCP_NODELAY`即可，如下所示："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "  /**\n",
    "   * @brief 禁用 Nagle 算法，降低 TCP 延迟。\n",
    "   *\n",
    "   * @param fd TCP socket 的文件描述符。\n",
    "   * @return 设置成功返回 true。\n",
    "   */\n",
    "  inline auto disableNagle(int fd) -> bool {\n",
    "    int one = 1;\n",
    "    return (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, reinterpret_cast<void *>(&one), sizeof(one)) != -1);\n",
    "  }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "在下一节中，在最终实现创建套接字的功能之前，我们将定义更多函数来设置可选和/或附加功能。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 设置其他参数\n",
    "\n",
    "首先，我们定义一个简单的方法来检查套接字操作是否会阻塞。这只是将全局错误变量`errno`与两个可能的值`EWOULDBLOCK`和`EINPROGRESS`进行比较："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// 版本变更, 代码不存在\n",
    "\n",
    "auto wouldBlock() -> bool {\n",
    "    return (errno == EWOULDBLOCK || errno == EINPROGRESS);\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "接下来，我们定义一个方法来设置套接字的生存时间（TTL，Time to Live ）值。TTL是一个网络层设置，用于控制数据包从发送方到接收方可以经过的最大跳数。根据应用程序的设置，我们可能并不需要使用它，但如果有需要，这里提供了该功能。同样幸运的是，我们所要做的就是使用`setsockopt()`函数，为非多播套接字设置`IP_TTL`套接字选项，为多播套接字设置`IP_MULTICAST_TTL`选项，如下所示："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// 版本变更, 代码不存在\n",
    "\n",
    "auto setTTL(int fd, int ttl) -> bool {\n",
    "    return (setsockopt(fd, IPPROTO_IP, IP_TTL,\n",
    "                       reinterpret_cast<void *>(&ttl), sizeof(ttl)) != -1);\n",
    "}\n",
    "auto setMcastTTL(int fd, int mcast_ttl) noexcept -> bool {\n",
    "    return (setsockopt(fd, IPPROTO_IP, IP_MULTICAST_TTL,\n",
    "                       reinterpret_cast<void *>(&mcast_ttl), sizeof(mcast_ttl)) != -1);\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "最后，我们定义最后一个方法，该方法允许我们在网络数据包到达网络套接字时生成软件时间戳。请注意，如果我们有支持硬件时间戳的专用硬件（网络接口卡，NICs），我们会在这里启用并使用它们。然而，为了限制本书的范围，我们假设你没有任何特殊硬件，只能使用`setsockopt()`方法设置`SO_TIMESTAMP`选项来启用软件时间戳："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "  /**\n",
    "   * @brief 开启软件接收时间戳（SO_TIMESTAMP）。\n",
    "   *\n",
    "   * @param fd socket 文件描述符。\n",
    "   * @return 设置成功返回 true。\n",
    "   *\n",
    "   * @note\n",
    "   *  - 若配合 Logger，可在日志里记录出错的 errno，辅助定位平台兼容性问题。\n",
    "   */\n",
    "  inline auto setSOTimestamp(int fd) -> bool {\n",
    "    int one = 1;\n",
    "    return (setsockopt(fd, SOL_SOCKET, SO_TIMESTAMP, reinterpret_cast<void *>(&one), sizeof(one)) != -1);\n",
    "  }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "关于套接字相关的实用函数的讨论到此结束，现在我们终于可以继续实现创建通用Unix套接字的功能了。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 创建套接字"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "在`createSocket()`方法的第一部分，我们首先检查是否提供了非空的`t_ip`，它代表接口IP，例如`192.168.10.104`。如果没有提供，我们将使用之前构建的`getIfaceIP()`方法从提供的接口名称中获取一个IP。我们还需要根据传入的参数填充`addrinfo`结构体，因为我们需要将其传递给`getaddrinfo()`函数，该函数将返回一个链表，最终用于构建实际的套接字。请注意，在`createSocket()`方法中，一旦创建套接字或用正确的参数初始化套接字失败，我们就返回`-1`表示失败："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "  /**\n",
    "   * @brief 创建 TCP/UDP socket，可用于客户端连接或服务端监听。\n",
    "   *\n",
    "   * @param logger     Common::Logger 实例，负责异步记录创建过程。（依赖 logging.h，\n",
    "   *                   Logger 内部通过 LFQueue + createAndStartThread 实现异步落盘。）\n",
    "   * @param socket_cfg socket 配置项，包含接口/IP/端口等信息。\n",
    "   * @return 成功返回 socket 文件描述符；若中途 ASSERT 失败会触发 FATAL 终止进程。\n",
    "   *\n",
    "   * @details\n",
    "   *  1. 若未显式指定 IP，会通过 getIfaceIP() 基于 iface 字段查询，结果写入 Logger。\n",
    "   *  2. 根据 is_udp_ / is_listening_ 组合，分别执行 connect()、bind()、listen() 等操作。\n",
    "   *  3. 默认把 socket 置为非阻塞，并在 TCP 场景下禁用 Nagle。\n",
    "   *  4. 若 needs_so_timestamp_ 为真，则开启 SO_TIMESTAMP。\n",
    "   *\n",
    "   * @warning\n",
    "   *  - 函数内部大量使用 ASSERT 宏（来自 macros.h）。配合 Logger 时，可在捕获到异常后\n",
    "   *    从日志文件中还原调用参数。\n",
    "   *  - createSocket 只负责创建与基础配置，不会注册 epoll 或绑定线程亲和性。在高性能\n",
    "   *    场景里，可结合 thread_utils.hpp 提供的 setThreadCore 在外部为 I/O 线程设置\n",
    "   *    CPU 亲和。\n",
    "   */\n",
    "[[nodiscard]] inline auto createSocket(Logger &logger, const SocketCfg& socket_cfg) -> int {\n",
    "    // 用于缓存 getCurrentTimeStr() 的返回值，便于 Logger 异步写盘时记录时间戳。\n",
    "    std::string time_str;\n",
    "\n",
    "    // 如果配置中未显式给出 IP，则根据网卡名查询；结果同样写入日志，方便排查。\n",
    "    const auto ip = socket_cfg.ip_.empty() ? getIfaceIP(socket_cfg.iface_) : socket_cfg.ip_;\n",
    "    logger.log(\"%:% %() % cfg:%\\n\", __FILE__, __LINE__, __FUNCTION__,\n",
    "               Common::getCurrentTimeStr(&time_str), socket_cfg.toString());\n",
    "\n",
    "    // 组合 AI_PASSIVE / AI_NUMERICHOST / AI_NUMERICSERV 等标志，为 getaddrinfo 做准备。\n",
    "    const int input_flags = (socket_cfg.is_listening_ ? AI_PASSIVE : 0) | (AI_NUMERICHOST | AI_NUMERICSERV);\n",
    "    // 根据协议类型填充 addrinfo 提示结构，针对 UDP/TCP 选择不同的 socktype / protocol。\n",
    "    const addrinfo hints{input_flags, AF_INET, socket_cfg.is_udp_ ? SOCK_DGRAM : SOCK_STREAM,\n",
    "                         socket_cfg.is_udp_ ? IPPROTO_UDP : IPPROTO_TCP, 0, 0, nullptr, nullptr};\n",
    "\n",
    "    addrinfo *result = nullptr;\n",
    "    // 解析 IP:Port -> addrinfo 链表。失败直接 ASSERT（会触发 FATAL）。\n",
    "    const auto rc = getaddrinfo(ip.c_str(), std::to_string(socket_cfg.port_).c_str(), &hints, &result);\n",
    "    ASSERT(!rc, \"getaddrinfo() failed. error:\" + std::string(gai_strerror(rc)) + \"errno:\" + strerror(errno));\n",
    "    // ...\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "接下来的部分检查传递给`createSocket()`方法的参数，并使用我们之前构建的所有方法根据需要设置正确的套接字参数。请注意，我们使用`getaddrinfo()`返回的`addrinfo * result`对象，通过`socket()`函数创建套接字。\n",
    "\n",
    "首先，我们实际调用函数来创建套接字："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "[[nodiscard]] inline auto createSocket(Logger &logger, const SocketCfg& socket_cfg) -> int {\n",
    "    // ...\n",
    "\n",
    "    int socket_fd = -1;\n",
    "    int one = 1;\n",
    "    // 遍历 getaddrinfo 返回的多个候选地址，逐一尝试创建/配置 socket。\n",
    "    for (addrinfo *rp = result; rp; rp = rp->ai_next) {\n",
    "      // 创建 socket，失败则记录 errno 并触发断言。\n",
    "      ASSERT((socket_fd = socket(rp->ai_family, rp->ai_socktype, rp->ai_protocol)) != -1,\n",
    "             \"socket() failed. errno:\" + std::string(strerror(errno)));\n",
    "    // ...\n",
    "    }\n",
    "\n",
    "    // ...\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "接下来，我们使用之前定义的方法将其设置为非阻塞模式，并禁用Nagle算法："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "      // 设置成非阻塞模式，避免阻塞 I/O。失败同样 ASSERT。\n",
    "      ASSERT(setNonBlocking(socket_fd), \"setNonBlocking() failed. errno:\" + std::string(strerror(errno)));\n",
    "\n",
    "      if (!socket_cfg.is_udp_) { // TCP 场景下关闭 Nagle，以降低小包延迟。\n",
    "        ASSERT(disableNagle(socket_fd), \"disableNagle() failed. errno:\" + std::string(strerror(errno)));\n",
    "      }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "接下来，如果它不是监听套接字，我们将套接字连接到目标地址："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "      if (!socket_cfg.is_listening_) { // 客户端模式：直接 connect 到目标地址。\n",
    "        ASSERT(connect(socket_fd, rp->ai_addr, rp->ai_addrlen) != 1,\n",
    "               \"connect() failed. errno:\" + std::string(strerror(errno)));\n",
    "      }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "然后，如果我们想要创建一个监听传入连接的套接字，我们需要设置正确的参数，并将套接字绑定到客户端尝试连接的特定地址。对于这种套接字配置，我们还需要调用`listen()`函数。请注意，这里我们引用了`MaxTCPServerBacklog`参数，其定义如下："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "  /// 最大的监听队列长度。若使用 Logger 记录相关事件，可将该值一并写入日志便于排查。\n",
    "  constexpr int MaxTCPServerBacklog = 1024;"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "现在，让我们看看将套接字设置为监听套接字的代码："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "      if (socket_cfg.is_listening_) {\n",
    "        // 绑定端口（UDP 直接绑定 INADDR_ANY，TCP 使用 getaddrinfo 给出的地址）。\n",
    "        const sockaddr_in addr{AF_INET, htons(socket_cfg.port_), {htonl(INADDR_ANY)}, {}};\n",
    "        ASSERT(bind(socket_fd,\n",
    "                    socket_cfg.is_udp_ ? reinterpret_cast<const struct sockaddr *>(&addr) : rp->ai_addr,\n",
    "                    sizeof(addr)) == 0,\n",
    "               \"bind() failed. errno:%\" + std::string(strerror(errno)));\n",
    "      }\n",
    "\n",
    "      if (!socket_cfg.is_udp_ && socket_cfg.is_listening_) { // TCP 服务端：进入监听状态。\n",
    "        ASSERT(listen(socket_fd, MaxTCPServerBacklog) == 0,\n",
    "               \"listen() failed. errno:\" + std::string(strerror(errno)));\n",
    "      }\n",
    "\n",
    "      if (socket_cfg.needs_so_timestamp_) { // 若需要接收时间戳，开启 SO_TIMESTAMP。\n",
    "        ASSERT(setSOTimestamp(socket_fd), \"setSOTimestamp() failed. errno:\" + std::string(strerror(errno)));\n",
    "      }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "最后，我们为刚创建的套接字设置TTL值并返回该套接字。我们还将使用之前创建的`setSOTimestamp()`方法，设置获取传入数据包数据接收时间戳的功能："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// 版本变更, 代码不存在\n",
    "\n",
    "if (is_udp && ttl) {\n",
    "    const bool is_multicast = atoi(ip.c_str()) & 0xe0;\n",
    "    if (is_multicast &&!setMcastTTL(fd, ttl)) {\n",
    "        logger.log(\"setMcastTTL() failed. errno:%\\n\",\n",
    "                   strerror(errno));\n",
    "        return -1;\n",
    "    }\n",
    "    if (!is_multicast &&!setTTL(fd, ttl)) {\n",
    "        logger.log(\"setTTL() failed. errno:%\\n\",\n",
    "                   strerror(errno));\n",
    "        return -1;\n",
    "    }\n",
    "}\n",
    "if (needs_so_timestamp &&!setSOTimestamp(fd)) {\n",
    "    logger.log(\"setSOTimestamp() failed. errno:%\\n\",\n",
    "               strerror(errno));\n",
    "    return -1;\n",
    "}\n",
    "if (result)\n",
    "    freeaddrinfo(result);\n",
    "return fd;"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "既然我们已经讨论并实现了底层套接字方法的细节，就可以进入下一节，在这个方法的基础上构建一个更高层次的抽象。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 实现发送/接收TCP套接字\n",
    "\n",
    "既然我们已经完成了创建套接字并设置不同参数的基本方法的设计与实现，就可以开始使用它们了。首先，我们将实现一个`TCPSocket`结构体，它基于上一节创建的套接字实用工具。`TCPSocket`可用于发送和接收数据，因此它将在TCP套接字服务器和客户端中都得到应用。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 定义TCP套接字的数据成员\n",
    "\n",
    "让我们开始实现`TCPSocket`结构体，从所需的数据成员开始。由于这个套接字将用于发送和接收数据，我们将创建两个缓冲区，一个用于存储要发送的数据，另一个用于存储刚刚读取的数据。我们还将把与TCP套接字对应的文件描述符存储在`fd_`变量中。我们创建两个标志，一个用于跟踪发送套接字是否已连接，另一个用于检查接收套接字是否已连接。我们还保存一个对`Logger`对象的引用，纯粹用于日志记录目的。最后，我们存储一个`std::function`对象，当有新数据可供消费时，我们将使用它向想要从这个套接字读取数据的组件分发回调。本节的代码位于本书GitHub仓库的[`Chapter4/tcp_socket.h`](https://github.com/PacktPublishing/Building-Low-Latency-Applications-with-CPP/blob/main/Chapter4/tcp_server.h)和[`Chapter4/tcp_socket.cpp`](https://github.com/PacktPublishing/Building-Low-Latency-Applications-with-CPP/blob/main/Chapter4/tcp_server.cpp)中："
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- [tcp_socket.h](./libcommon/tcp_socket.h)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "我们定义一个默认的接收回调函数，用于初始化`recv_callback_`数据成员。这个方法只是记录确认回调已被调用的信息："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// 版本变更, 代码不存在\n",
    "\n",
    "auto defaultRecvCallback(TCPSocket *socket, Nanos rx_time) noexcept {\n",
    "    logger_.log(\"%:% %() % TCPSocket::defaultRecvCallback() socket:% len:% rx:%\\n\",\n",
    "                __FILE__, __LINE__, __FUNCTION__,\n",
    "                Common::getCurrentTimeStr(&time_str_),\n",
    "                socket->fd_, socket->next_rcv_valid_index_, rx_time);\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "接下来，让我们看看`TCPSocket`结构体的构造函数。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 构造和销毁TCP套接字"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "对于构造函数，我们将在堆上创建`send_buffer_`和`rcv_buffer_`字符指针存储，并通过一个lambda方法将`defaultRecvCallback()`方法赋值给`recv_callback_`成员变量。请注意，我们将套接字的接收和发送缓冲区大小设置为`TCPBufferSize`，定义如下："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "  /// 发送/接收缓冲区的固定大小（64 MB）——需要根据业务吞吐量评估是否足够。\n",
    "  constexpr size_t TCPBufferSize = 64 * 1024 * 1024;  \n",
    "\n",
    "  /**\n",
    "   * @brief TCP socket 封装，负责 socket 的生命周期、缓冲区管理与回调触发。\n",
    "   *\n",
    "   * Logger 引用被显式传入，不允许默认构造/拷贝/移动，确保后台日志线程与队列\n",
    "   * 的资源不会被多个实例共享导致竞态。\n",
    "   */\n",
    "  struct TCPSocket {\n",
    "    explicit TCPSocket(Logger &logger)\n",
    "        : logger_(logger) {\n",
    "      // 初始化收发缓冲区，避免运行时频繁扩容造成抖动。\n",
    "      outbound_data_.resize(TCPBufferSize);\n",
    "      inbound_data_.resize(TCPBufferSize);\n",
    "    }\n",
    "    // ...\n",
    "  }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "然后，我们创建`destroy()`函数和析构函数来执行简单的清理任务。我们将关闭套接字文件描述符，并销毁在构造函数中创建的接收和发送缓冲区："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// 版本变更, 代码不存在\n",
    "\n",
    "auto TCPSocket::destroy() noexcept -> void {\n",
    "    close(fd_);\n",
    "    fd_ = -1;\n",
    "}\n",
    "~TCPSocket() {\n",
    "    destroy();\n",
    "    delete[] send_buffer_;\n",
    "    send_buffer_ = nullptr;\n",
    "    delete[] rcv_buffer_;\n",
    "    rcv_buffer_ = nullptr;\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "我们定义之前见过的样板代码，以防止意外或无意的构造、复制或赋值："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "    /// 禁用默认构造与拷贝/移动，防止 Logger & socket 资源被不当共享。\n",
    "    TCPSocket() = delete;\n",
    "    TCPSocket(const TCPSocket &) = delete;\n",
    "    TCPSocket(const TCPSocket &&) = delete;\n",
    "    TCPSocket &operator=(const TCPSocket &) = delete;\n",
    "    TCPSocket &operator=(const TCPSocket &&) = delete;"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "接下来，让我们尝试对这个套接字执行一个关键操作——建立TCP连接。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 建立TCP连接"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "对于这个结构体，我们将定义一个`connect()`方法，该方法主要用于创建、初始化并连接`TCPSocket`。我们将使用上一节创建的`createSocket()`方法，并传入正确的参数来实现这一功能："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "  /**\n",
    "   * @brief 根据配置连接或监听 TCP socket。\n",
    "   *\n",
    "   * @param ip           目标 IP（监听模式可留空）。\n",
    "   * @param iface        网络接口名（监听模式需指定）。\n",
    "   * @param port         目标端口。\n",
    "   * @param is_listening 是否作为监听端口。\n",
    "   * @return 返回创建的 socket 描述符；失败时触发 ASSERT 并终止。\n",
    "   *\n",
    "   * @note 底层调用 createSocket()，该函数会通过 Logger 异步输出详细参数与错误。\n",
    "   */  \n",
    "/// Create TCPSocket with provided attributes to either listen-on / connect-to.\n",
    "  auto TCPSocket::connect(const std::string &ip, const std::string &iface, int port, bool is_listening) -> int {\n",
    "    // 对于 FIFOSequencer 等组件，需要抓取内核时间戳，因此 needs_so_timestamp=true。\n",
    "    const SocketCfg socket_cfg{ip, iface, port, false, is_listening, true};\n",
    "    socket_fd_ = createSocket(logger_, socket_cfg);  // createSocket 会通过 Logger 记录调用细节。\n",
    "\n",
    "    // 存储远端地址（监听场景下为本地 INADDR_ANY/port），后续 recvmsg/send 需要用到。\n",
    "    socket_attrib_.sin_addr.s_addr = INADDR_ANY;\n",
    "    socket_attrib_.sin_port = htons(port);\n",
    "    socket_attrib_.sin_family = AF_INET;\n",
    "\n",
    "    return socket_fd_;\n",
    "  }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "接下来，我们将继续探讨套接字的下一个关键功能——发送和接收数据。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 发送和接收数据"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "我们在讨论中提到，当有新数据可用时，感兴趣的监听器将通过`recv_callback_`这个`std::function`机制得到通知。因此，我们只需要为这个结构体的用户提供一个`send()`方法，用于发送数据。请注意，这个`send()`方法只是将提供的数据复制到发送缓冲区中，实际的网络写入操作将在我们稍后会看到的`sendAndRecv()`方法中完成："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "  /**\n",
    "   * @brief 将上层待发送的数据写入发送缓冲区。\n",
    "   *\n",
    "   * @param data 指向待发送内容。\n",
    "   * @param len  数据长度。\n",
    "   *\n",
    "   * @note 只是拷贝到 outbound buffer；真正的发送在 sendAndRecv() 中完成。\n",
    "   */\n",
    "  /// Write outgoing data to the send buffers.\n",
    "  auto TCPSocket::send(const void *data, size_t len) noexcept -> void {\n",
    "    // 将数据拷贝入发送缓冲区，等待 sendAndRecv() 统一发送。\n",
    "    memcpy(outbound_data_.data() + next_send_valid_index_, data, len);\n",
    "    next_send_valid_index_ += len;\n",
    "  }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "最后，`TCPSocket`结构体中最重要的方法是`sendAndRecv()`，它将可用数据读入`rcv_buffer_`，增加计数器，如果读取到了一定数量的数据，还会调度`recv_callback_`。该方法的后半部分则执行相反的操作——它尝试使用`send()`函数将`send_buffer_`中的数据写出，并更新索引跟踪变量："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "  /**\n",
    "   * @brief 非阻塞地发送/接收数据，必要时触发回调。\n",
    "   *\n",
    "   * @return 若本次调用读取到了数据返回 true，否则返回 false。\n",
    "   *\n",
    "   * @details\n",
    "   *  - 首先尝试读取（recvmsg），处理内核时间戳并将信息写入日志。\n",
    "   *  - 若有待发送数据，则调用 send() 将缓冲区内容写出去，并记录日志。\n",
    "   *  - 典型场景：由 IO 线程在 epoll 事件后调度。\n",
    "   */\n",
    "  /// Called to publish outgoing data from the buffers as well as check for and callback if data is available in the read buffers.\n",
    "  auto TCPSocket::sendAndRecv() noexcept -> bool {\n",
    "    // ctrl 缓冲区用于接收 SCM_TIMESTAMP 控制信息。\n",
    "    char ctrl[CMSG_SPACE(sizeof(struct timeval))];\n",
    "    auto cmsg = reinterpret_cast<struct cmsghdr *>(&ctrl);\n",
    "\n",
    "    // iovec 指向 inbound buffer 中尚未填充的空间，recvmsg 直接写入。\n",
    "    iovec iov{inbound_data_.data() + next_rcv_valid_index_, TCPBufferSize - next_rcv_valid_index_};\n",
    "    msghdr msg{&socket_attrib_, sizeof(socket_attrib_), &iov, 1, ctrl, sizeof(ctrl), 0};\n",
    "\n",
    "    // 非阻塞读取，可同时获取控制信息（内核时间戳等）。\n",
    "    const auto read_size = recvmsg(socket_fd_, &msg, MSG_DONTWAIT);\n",
    "    if (read_size > 0) {\n",
    "      next_rcv_valid_index_ += read_size;  // 更新有效数据长度，供上层解析。\n",
    "\n",
    "      Nanos kernel_time = 0;\n",
    "      timeval time_kernel;\n",
    "      if (cmsg->cmsg_level == SOL_SOCKET &&\n",
    "          cmsg->cmsg_type == SCM_TIMESTAMP &&\n",
    "          cmsg->cmsg_len == CMSG_LEN(sizeof(time_kernel))) {\n",
    "        memcpy(&time_kernel, CMSG_DATA(cmsg), sizeof(time_kernel));\n",
    "        // 将 timeval 转换为纳秒，方便与用户时间戳比较。\n",
    "        kernel_time = time_kernel.tv_sec * NANOS_TO_SECS + time_kernel.tv_usec * NANOS_TO_MICROS;\n",
    "      }\n",
    "\n",
    "      const auto user_time = getCurrentNanos();  // 用户态当前时间戳。\n",
    "\n",
    "      // 通过 Logger 记录一次收包：包括 socket fd / 长度 / 用户与内核时间差。\n",
    "      logger_.log(\"%:% %() % read socket:% len:% utime:% ktime:% diff:%\\n\", __FILE__, __LINE__, __FUNCTION__,\n",
    "                  Common::getCurrentTimeStr(&time_str_), socket_fd_, next_rcv_valid_index_, user_time, kernel_time, (user_time - kernel_time));\n",
    "      // 调用上层回调，交由业务线程处理数据；注意回调在网络线程执行。\n",
    "      recv_callback_(this, kernel_time);\n",
    "    }\n",
    "\n",
    "    if (next_send_valid_index_ > 0) {\n",
    "      // 非阻塞发送缓冲区数据，使用 MSG_NOSIGNAL 防止对端关闭导致 SIGPIPE。\n",
    "      const auto n = ::send(socket_fd_, outbound_data_.data(), next_send_valid_index_, MSG_DONTWAIT | MSG_NOSIGNAL);\n",
    "      logger_.log(\"%:% %() % send socket:% len:%\\n\", __FILE__, __LINE__, __FUNCTION__,\n",
    "                  Common::getCurrentTimeStr(&time_str_), socket_fd_, n);\n",
    "    }\n",
    "    next_send_valid_index_ = 0;  // 无论成功与否都重置游标；失败重试由上层控制。\n",
    "\n",
    "    return (read_size > 0);\n",
    "  }\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "关于`TCPSocket`类的讨论到此结束。接下来，我们将构建一个类，用于封装和管理`TCPSocket`对象。它将用于在充当服务器的组件中实现TCP服务器的功能。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 构建TCP服务器组件\n",
    "\n",
    "在上一节中，我们构建了`TCPSocket`类，需要连接到TCP连接并进行数据收发的组件可以使用这个类。在本节中，我们将构建一个`TCPServer`组件，它在内部管理多个这样的`TCPSocket`对象。它还管理诸如监听、接受和跟踪新的传入连接，以及在这些套接字集合上进行数据收发等任务。`TCPServer`组件的所有源代码都在本书GitHub仓库中的[`Chapter4/tcp_server.h`](https://github.com/PacktPublishing/Building-Low-Latency-Applications-with-CPP/blob/main/Chapter7/common/tcp_server.h)和[`Chapter4/tcp_server.cpp`](https://github.com/PacktPublishing/Building-Low-Latency-Applications-with-CPP/blob/main/Chapter7/common/tcp_server.cpp)文件中。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 定义TCP服务器的数据成员"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "首先，我们将定义并描述`TCPServer`类包含的数据成员。它需要一个文件描述符`efd_`和一个对应的`TCPSocket`对象`listener_socket_`，用于表示监听客户端新传入连接的套接字。它维护一个`epoll_event`数组`events_`，用于监控监听套接字文件描述符以及已连接客户端的套接字描述符。它有几个`std::vector`类型的套接字对象——我们期望接收数据的套接字、期望发送数据的套接字以及已断开连接的套接字。我们很快就会看到这些是如何使用的。\n",
    "\n",
    "这个类有两个`std::function`对象——一个用于在接收到新数据时调度回调，另一个在当前一轮套接字轮询完成所有回调后调度。为了更好地解释这一点，我们首先使用`epoll`调用找到所有有数据可读的套接字，为每个有数据的套接字调度`recv_callback_`，最后，当所有套接字都被通知后，调度`recv_finished_callback_`。这里需要注意的另一点是，`recv_callback_`提供了接收到数据的`TCPSocket`，以及`Nanos`类型的`rx_time`，用于指定该套接字上数据的软件接收时间。接收时间戳用于按照接收顺序精确处理TCP数据包，因为TCP服务器会监控并从许多不同的客户端TCP套接字读取数据："
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- [tcp_server.h](./libcommon/tcp_server.h)\n",
    "- [tcp_server.cpp](./libcommon/tcp_server.cpp)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "在下一节中，我们将查看初始化这些字段以及销毁TCP服务器的代码。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 初始化和销毁TCP服务器"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`TCPServer`的构造函数很直接——它初始化`listener_socket_`和`logger_`，并设置默认的回调接收器，就像我们对`TCPSocket`所做的那样："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "  /**\n",
    "   * @brief TCPServer 封装：管理监听 socket、epoll 事件、连接集合与回调。\n",
    "   *\n",
    "   * Logger 在整个服务器生命周期中同步输出事件，以便通过异步日志线程分析连接状态。\n",
    "   */\n",
    "  struct TCPServer {\n",
    "    explicit TCPServer(Logger &logger)\n",
    "        : listener_socket_(logger), logger_(logger) {\n",
    "    }\n",
    "    // ...\n",
    "  };"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "我们在这里定义默认的接收回调方法，这些方法除了记录接收到回调外什么也不做。反正这些只是占位符，因为在实际应用中我们会设置不同的回调："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// 版本变更, 代码不存在\n",
    "\n",
    "auto defaultRecvCallback(TCPSocket *socket, Nanos rx_time) noexcept {\n",
    "    logger_.log(\"%:% %() % TCPServer::defaultRecvCallback() socket:% len:% rx:%\\n\", __FILE__, __LINE__, __FUNCTION__,\n",
    "                Common::getCurrentTimeStr(&time_str_), socket->fd_, socket->next_rcv_valid_index_, rx_time);\n",
    "}\n",
    "auto defaultRecvFinishedCallback() noexcept {\n",
    "    logger_.log(\"%:% %() % TCPServer::defaultRecvFinishedCallback()\\n\", __FILE__, __LINE__, __FUNCTION__,\n",
    "                Common::getCurrentTimeStr(&time_str_));\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "销毁套接字的代码也很简单——我们关闭文件描述符并销毁`TCPSocket`对象`listener_socket_`："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// 版本变更, 代码不存在\n",
    "\n",
    "auto TCPServer::destroy() {\n",
    "    close(efd_);\n",
    "    efd_ = -1;\n",
    "    listener_socket_.destroy();\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "最后，我们给出之前见过的这个类的模板代码："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// 版本变更, 代码不存在\n",
    "\n",
    "TCPServer() = delete;\n",
    "TCPServer(const TCPServer &) = delete;\n",
    "TCPServer(const TCPServer &&) = delete;\n",
    "TCPServer &operator=(const TCPServer &) = delete;\n",
    "TCPServer &operator=(const TCPServer &&) = delete;"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "接下来，让我们了解一下初始化监听套接字的代码。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 启动并监听新连接"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`TCPServer::listen()`方法首先使用`epoll_create()`这个Linux系统调用创建一个新的`epoll`实例，并将其保存到`efd_`变量中。它使用我们之前构建的`TCPSocket::connect()`方法来初始化`listener_socket_`，但这里重要的是，我们将监听参数设置为`true`。最后，我们使用`epoll_add()`方法将`listener_socket_`添加到要监控的套接字列表中，因为最初这是唯一要监控的套接字。我们将在下一节查看这个`epoll_add()`方法："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "  /**\n",
    "   * @brief 在指定接口/端口上启动监听。\n",
    "   *\n",
    "   * @param iface 网络接口名（如 \"eth0\"）。\n",
    "   * @param port  监听端口。\n",
    "   *\n",
    "   * @note 内部创建 epoll fd 并将监听 socket 加入事件列表。\n",
    "   */\n",
    "  /// Start listening for connections on the provided interface and port.\n",
    "  auto TCPServer::listen(const std::string &iface, int port) -> void {\n",
    "    epoll_fd_ = epoll_create(1);\n",
    "    ASSERT(epoll_fd_ >= 0, \"epoll_create() failed error:\" + std::string(std::strerror(errno)));\n",
    "\n",
    "    // 创建监听 socket；connect() 会通过 Logger 记录底层细节。\n",
    "    ASSERT(listener_socket_.connect(\"\", iface, port, true) >= 0,\n",
    "           \"Listener socket failed to connect. iface:\" + iface + \" port:\" + std::to_string(port) + \" error:\" +\n",
    "           std::string(std::strerror(errno)));\n",
    "\n",
    "    // 将监听 fd 加入 epoll，之后 poll() 才能收到新连接事件。\n",
    "    ASSERT(addToEpollList(&listener_socket_), \"epoll_ctl() failed. error:\" + std::string(std::strerror(errno)));\n",
    "  }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "现在，让我们在下一小节中看看`epoll_add()`以及与之互补的`epoll_del()`方法是如何构建的。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 添加和移除监控套接字"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`epoll_add()`方法用于将`TCPSocket`添加到要监控的套接字列表中。它使用`epoll_ctl()`系统调用，并传入`EPOLL_CTL_ADD`参数，将提供的套接字文件描述符添加到`efd_`这个`epoll`类成员中。启用`EPOLLET`表示开启边缘触发（edge-triggered）的`epoll`选项，简单来说，这意味着当有数据需要读取时，你只会收到一次通知，而不是持续收到提醒。在这种模式下，由应用程序开发者自行决定何时读取数据。`EPOLLIN`用于在有数据可读时发出通知："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// 版本变更, 代码不存在\n",
    "\n",
    "auto TCPServer::epoll_add(TCPSocket *socket) {\n",
    "    epoll_event ev{};\n",
    "    ev.events = EPOLLET | EPOLLIN;\n",
    "    ev.data.ptr = reinterpret_cast<void *>(socket);\n",
    "    return (epoll_ctl(efd_, EPOLL_CTL_ADD, socket->fd_, &ev) != -1);\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "  /**\n",
    "   * @brief 将 socket 添加到 epoll 列表（边缘触发 + EPOLLIN）。\n",
    "   *\n",
    "   * @param socket 待添加的 TCPSocket。\n",
    "   * @return 成功返回 true（内部使用 epoll_ctl）。\n",
    "   */  \n",
    "  /// Add and remove socket file descriptors to and from the EPOLL list.\n",
    "  auto TCPServer::addToEpollList(TCPSocket *socket) {\n",
    "    // 边缘触发 + 可读事件，将 socket 指针存入 event data 以便回调时复原。\n",
    "    epoll_event ev{EPOLLET | EPOLLIN, {reinterpret_cast<void *>(socket)}};\n",
    "    return !epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, socket->socket_fd_, &ev);\n",
    "  }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`epoll_del()`的功能与`epoll_add()`相反——仍然使用`epoll_ctl()`，但这次使用`EPOLL_CTL_DEL`参数，将`TCPSocket`从监控的套接字列表中移除："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// 版本变更, 代码不存在\n",
    "\n",
    "auto TCPServer::epoll_del(TCPSocket *socket) {\n",
    "    return (epoll_ctl(efd_, EPOLL_CTL_DEL, socket->fd_, nullptr) != -1);\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "我们在这里构建的`del()`方法，会将`TCPSocket`从监控的套接字列表中移除，同时也会从套接字的不同数据成员容器中移除："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// 版本变更, 代码不存在\n",
    "\n",
    "auto TCPServer::del(TCPSocket *socket) {\n",
    "    epoll_del(socket);\n",
    "    sockets_.erase(std::remove(sockets_.begin(), sockets_.end(), socket), sockets_.end());\n",
    "    receive_sockets_.erase(std::remove(receive_sockets_.begin(), receive_sockets_.end(), socket), receive_sockets_.end());\n",
    "    send_sockets_.erase(std::remove(send_sockets_.begin(), send_sockets_.end(), socket), send_sockets_.end());\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "现在，我们来看这部分最重要的方法——`TCPServer::poll()`，它将用于执行以下几项任务：\n",
    "\n",
    "- 调用`epoll_wait()`，检测是否有新的传入连接，如果有，则将其添加到我们的容器中。\n",
    "- 从`epoll_wait()`的调用结果中，检测从客户端断开连接的套接字，并将它们从我们的容器中移除。\n",
    "- 从`epoll_wait()`的调用结果中，检查是否有套接字有数据准备好读取或有待发送的数据。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "让我们将整个方法分成几个部分——首先是调用`epoll_wait()`方法的部分，传入`epoll`实例和最大事件数，最大事件数为我们容器中套接字的总数，并且不设置超时："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "  /**\n",
    "   * @brief 轮询 epoll 事件，维护 socket 容器。\n",
    "   *\n",
    "   * @note 将新连接/可读/可写的 socket 添加到 receive/send 队列，并在日志中记录。\n",
    "   */\n",
    "  /// Check for new connections or dead connections and update containers that track the sockets.\n",
    "  auto TCPServer::poll() noexcept -> void {\n",
    "    const int max_events = 1 + send_sockets_.size() + receive_sockets_.size();\n",
    "\n",
    "    // epoll_wait 非阻塞（timeout=0），返回当前有事件的数量。\n",
    "    const int n = epoll_wait(epoll_fd_, events_, max_events, 0);\n",
    "  // ...\n",
    "  }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "接下来，如果`epoll_wait()`的返回值`n`大于0，我们就遍历由它填充的`events_`数组。对于`events_`数组中的每个`epoll_event`，我们使用`event.data.ptr`对象，并将其转换为`TCPSocket*`，因为这是我们在`epoll_add()`方法中设置`events_`数组的方式："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "    bool have_new_connection = false;\n",
    "    for (int i = 0; i < n; ++i) {\n",
    "      const auto &event = events_[i];\n",
    "      auto socket = reinterpret_cast<TCPSocket *>(event.data.ptr);\n",
    "      // ...\n",
    "    }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "对于每个`epoll_event`条目，我们检查`events`标志中是否设置了`EPOLLIN`标志，这表示有一个新的套接字有数据可读。如果这个套接字恰好是`listener_socket_`，也就是我们配置用来监听连接的`TCPServer`主套接字，那么我们就知道有新连接需要添加。如果这是一个不同于`listener_socket_`的套接字，并且它不在`receive_sockets_`向量列表中，我们就将其添加到该列表中："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "      // 可读事件：可能是新连接（监听 socket），也可能是现有连接。\n",
    "      if (event.events & EPOLLIN) {\n",
    "        if (socket == &listener_socket_) {\n",
    "          logger_.log(\"%:% %() % EPOLLIN listener_socket:%\\n\", __FILE__, __LINE__, __FUNCTION__,\n",
    "                      Common::getCurrentTimeStr(&time_str_), socket->socket_fd_);\n",
    "          have_new_connection = true;\n",
    "          continue;\n",
    "        }\n",
    "        logger_.log(\"%:% %() % EPOLLIN socket:%\\n\", __FILE__, __LINE__, __FUNCTION__,\n",
    "                    Common::getCurrentTimeStr(&time_str_), socket->socket_fd_);\n",
    "        if (std::find(receive_sockets_.begin(), receive_sockets_.end(), socket) == receive_sockets_.end())\n",
    "          receive_sockets_.push_back(socket);\n",
    "      }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "类似地，我们检查`EPOLLOUT`标志，它表示有一个套接字可以用来发送数据，如果该套接字不在`send_sockets_`向量列表中，我们就将其添加进去："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "      // 可写事件：意味着 send buffer 有空间，可以尝试发送。\n",
    "      if (event.events & EPOLLOUT) {\n",
    "        logger_.log(\"%:% %() % EPOLLOUT socket:%\\n\", __FILE__, __LINE__, __FUNCTION__,\n",
    "                    Common::getCurrentTimeStr(&time_str_), socket->socket_fd_);\n",
    "        if (std::find(send_sockets_.begin(), send_sockets_.end(), socket) == send_sockets_.end())\n",
    "          send_sockets_.push_back(socket);\n",
    "      }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "最后，我们检查是否设置了`EPOLLERR`或`EPOLLHUP`标志，这两个标志分别表示出现错误或表示套接字已从另一端关闭（信号挂断）。在这种情况下，我们将这个套接字添加到`disconnected_sockets_`向量中，以便后续移除："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "      // 错误/挂起事件：通常也需要读一次以确认状态。\n",
    "      if (event.events & (EPOLLERR | EPOLLHUP)) {\n",
    "        logger_.log(\"%:% %() % EPOLLERR socket:%\\n\", __FILE__, __LINE__, __FUNCTION__,\n",
    "                    Common::getCurrentTimeStr(&time_str_), socket->socket_fd_);\n",
    "        if (std::find(receive_sockets_.begin(), receive_sockets_.end(), socket) == receive_sockets_.end())\n",
    "          receive_sockets_.push_back(socket);\n",
    "      }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "最后，在这个方法中，如果在上一个代码块中检测到有新连接，我们就需要接受这个新连接。我们使用`accept()`系统调用，并传入`listener_socket_`的文件描述符来实现这一点，获取这个新套接字的文件描述符。我们还使用之前构建的`setNonBlocking()`和`setNoDelay()`方法，将套接字设置为非阻塞模式并禁用Nagle算法："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "    // 如果有新连接到达，持续 accept 直到耗尽（因为是边缘触发）。\n",
    "    while (have_new_connection) {\n",
    "      logger_.log(\"%:% %() % have_new_connection\\n\", __FILE__, __LINE__, __FUNCTION__,\n",
    "                  Common::getCurrentTimeStr(&time_str_));\n",
    "      sockaddr_storage addr;\n",
    "      socklen_t addr_len = sizeof(addr);\n",
    "      int fd = accept(listener_socket_.socket_fd_, reinterpret_cast<sockaddr *>(&addr), &addr_len);\n",
    "      if (fd == -1)\n",
    "        break;  // 没有更多连接。\n",
    "\n",
    "      ASSERT(setNonBlocking(fd) && disableNagle(fd),\n",
    "             \"Failed to set non-blocking or no-delay on socket:\" + std::to_string(fd));\n",
    "\n",
    "      logger_.log(\"%:% %() % accepted socket:%\\n\", __FILE__, __LINE__, __FUNCTION__,\n",
    "                  Common::getCurrentTimeStr(&time_str_), fd);\n",
    "      // ...\n",
    "    }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "最后，我们使用这个文件描述符创建一个新的`TCPSocket`对象，并将这个`TCPSocket`对象添加到`sockets_`和`receive_sockets_`容器中："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "      auto socket = new TCPSocket(logger_);  // 为每个连接创建专属 TCPSocket。\n",
    "      socket->socket_fd_ = fd;\n",
    "      socket->recv_callback_ = recv_callback_;  // 继承服务器的收包回调。\n",
    "      ASSERT(addToEpollList(socket), \"Unable to add socket. error:\" + std::string(std::strerror(errno)));\n",
    "\n",
    "      if (std::find(receive_sockets_.begin(), receive_sockets_.end(), socket) == receive_sockets_.end())\n",
    "        receive_sockets_.push_back(socket);"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "到此，我们完成了寻找新连接、检测断开连接，以及监控现有连接以查看是否有数据可读所需的所有功能。下一小节将通过演示如何从有数据可读或待发送数据的套接字列表中收发数据，来完成对`TCPServer`类的介绍。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### 发送和接收数据"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "在有数据传入或传出的套接字列表上发送和接收数据的代码如下所示。实现过程非常直接，它只是对`receive_sockets_`和`send_sockets_`中的每个套接字调用`TCPSocket::sendAndRecv()`方法。对于传入的数据，调用`TCPSocket::sendAndRecv()`会触发`recv_callback_method`。这里我们需要做的一件事是检查这次是否读取到了任何数据，如果读取到了，在所有`recv_callback_`调用完成后，我们会触发`recv_finished_callback_`。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "  /**\n",
    "   * @brief 将所有收发 socket 执行一次 sendAndRecv()。\n",
    "   *\n",
    "   * @note\n",
    "   *  - 若任一 socket 有数据到达，会触发 recv_callback_，并在全部处理完毕后\n",
    "   *    调用 recv_finished_callback_ 通知监听者。\n",
    "   */\n",
    "  /// Publish outgoing data from the send buffer and read incoming data from the receive buffer.\n",
    "  auto TCPServer::sendAndRecv() noexcept -> void {\n",
    "    auto recv = false;\n",
    "\n",
    "    // 先处理所有可读 socket，若任意一个有数据到达，则 recv 置为 true。\n",
    "    std::for_each(receive_sockets_.begin(), receive_sockets_.end(), [&recv](auto socket) {\n",
    "      recv |= socket->sendAndRecv();\n",
    "    });\n",
    "\n",
    "    // 如果这一轮确实有数据到达并全部处理完，调用通知回调。\n",
    "    if (recv)\n",
    "      recv_finished_callback_();\n",
    "\n",
    "    // 再依次处理需要发送的 socket，触发 sendAndRecv() 内的非阻塞发送逻辑。\n",
    "    std::for_each(send_sockets_.begin(), send_sockets_.end(), [](auto socket) {\n",
    "      socket->sendAndRecv();\n",
    "    });\n",
    "  }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "至此，我们完成了`TCPServer`类的实现。让我们通过本节构建的所有内容的一个简单示例，来结束关于网络编程的讨论。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 构建TCP服务器和客户端示例"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "在本节中，我们将构建一个示例，并使用本节实现的`TCPSocket`和`TCPServer`类。这个示例可以在[`Chapter4/socket_example.cpp`](https://github.com/PacktPublishing/Building-Low-Latency-Applications-with-CPP/blob/main/Chapter7/common/socket_example.cpp)源文件中找到。这个简单的示例创建了一个`TCPServer`，它在`lo`接口（回环地址127.0.0.1）和监听端口12345上监听传入连接。`TCPServer`类通过`tcpServerRecvCallback()` lambda方法从连接的客户端接收数据，然后用一个简单的响应回传给客户端。然后，我们使用`TCPSocket`类创建五个客户端，每个客户端都连接到这个`TCPServer`。最后，每个客户端向服务器发送一些数据，服务器会返回响应，每个客户端反复调用`sendAndRecv()`来发送和接收数据。`TCPServer`调用`poll()`和`sendAndRecv()`来查找连接和数据并进行读取。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- [socket_example.cpp](./socket_example/socket_example.cpp)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "首先，设置回调lambda表达式的代码如下："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "  auto tcpServerRecvCallback = [&](TCPSocket *socket, Nanos rx_time) noexcept {\n",
    "    logger_.log(\"TCPServer::defaultRecvCallback() socket:% len:% rx:%\\n\",\n",
    "                socket->socket_fd_, socket->next_rcv_valid_index_, rx_time);\n",
    "\n",
    "    const std::string reply = \"TCPServer received msg:\" + std::string(socket->inbound_data_.data(), socket->next_rcv_valid_index_);\n",
    "    socket->next_rcv_valid_index_ = 0;\n",
    "\n",
    "    socket->send(reply.data(), reply.length());\n",
    "  };\n",
    "\n",
    "  auto tcpServerRecvFinishedCallback = [&]() noexcept {\n",
    "    logger_.log(\"TCPServer::defaultRecvFinishedCallback()\\n\");\n",
    "  };\n",
    "\n",
    "  auto tcpClientRecvCallback = [&](TCPSocket *socket, Nanos rx_time) noexcept {\n",
    "    const std::string recv_msg = std::string(socket->inbound_data_.data(), socket->next_rcv_valid_index_);\n",
    "    socket->next_rcv_valid_index_ = 0;\n",
    "\n",
    "    logger_.log(\"TCPSocket::defaultRecvCallback() socket:% len:% rx:% msg:%\\n\",\n",
    "                socket->socket_fd_, socket->next_rcv_valid_index_, rx_time, recv_msg);\n",
    "  };"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "然后，我们创建、初始化并连接服务器和客户端，如下所示："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "  const std::string iface = \"lo\";\n",
    "  const std::string ip = \"127.0.0.1\";\n",
    "  const int port = 12345;\n",
    "\n",
    "  logger_.log(\"Creating TCPServer on iface:% port:%\\n\", iface, port);\n",
    "  TCPServer server(logger_);\n",
    "  server.recv_callback_ = tcpServerRecvCallback;\n",
    "  server.recv_finished_callback_ = tcpServerRecvFinishedCallback;\n",
    "  server.listen(iface, port);\n",
    "\n",
    "  std::vector<TCPSocket *> clients(5);\n",
    "\n",
    "  for (size_t i = 0; i < clients.size(); ++i) {\n",
    "    clients[i] = new TCPSocket(logger_);\n",
    "    clients[i]->recv_callback_ = tcpClientRecvCallback;\n",
    "\n",
    "    logger_.log(\"Connecting TCPClient-[%] on ip:% iface:% port:%\\n\", i, ip, iface, port);\n",
    "    clients[i]->connect(ip, iface, port, false);\n",
    "    server.poll();\n",
    "  }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "最后，让客户端发送数据，并在客户端和服务器上调用相应的轮询以及发送/接收方法，如下所示："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "  using namespace std::literals::chrono_literals;\n",
    "\n",
    "  for (auto itr = 0; itr < 5; ++itr) {\n",
    "    for (size_t i = 0; i < clients.size(); ++i) {\n",
    "      const std::string client_msg = \"CLIENT-[\" + std::to_string(i) + \"] : Sending \" + std::to_string(itr * 100 + i);\n",
    "      logger_.log(\"Sending TCPClient-[%] %\\n\", i, client_msg);\n",
    "      clients[i]->send(client_msg.data(), client_msg.length());\n",
    "      clients[i]->sendAndRecv();\n",
    "\n",
    "      std::this_thread::sleep_for(500ms);\n",
    "      server.poll();\n",
    "      server.sendAndRecv();\n",
    "    }\n",
    "  }"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "运行这个示例，如下所示，日志文件中会输出类似的内容："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "! cd ../../build/Part_1/04/socket_example && cat socket_example.log"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "至此，我们关于使用套接字进行C++网络编程的讨论就结束了。我们涵盖了很多关于套接字编程的基本底层细节。我们还从服务器和客户端的角度，为TCP和UDP通信设计并实现了一些更高级别的抽象。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 总结"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "在本章中，我们踏入了低延迟应用C++开发的领域。我们构建了一些相对基础但极为有用的模块，这些模块可用于多种低延迟应用场景。我们将之前讨论的许多与有效使用C++和计算机体系结构特性相关的理论付诸实践，以构建低延迟且高性能的应用。\n",
    "\n",
    "第一个组件用于创建新的执行线程，并运行不同组件可能需要的函数。这里一个重要的功能是能够通过设置线程亲和力，控制新创建的线程绑定到哪个CPU核心。\n",
    "\n",
    "我们构建的第二个组件旨在避免在关键代码路径上进行动态内存分配。我们重申了动态内存分配存在的效率问题，并设计了一个内存池，在构造时从堆中预分配内存。然后，我们为该组件添加了实用功能，允许在运行时分配和释放对象，而无需依赖动态内存分配。\n",
    "\n",
    "接下来，我们构建了一个无锁的先进先出（First In First Out，FIFO）风格的队列，用于在单生产者单消费者（SPSC）场景下的线程间通信。这里的关键要求是，单个读取者和单个写入者能够在不使用任何锁或互斥量的情况下访问队列中的共享数据。没有锁和互斥量意味着没有上下文切换，正如前面所讨论的，上下文切换是多线程应用中效率低下和延迟的主要来源。\n",
    "\n",
    "我们列出的第四个组件是一个为对延迟敏感的应用提供高效日志记录的框架。日志记录是所有应用（包括低延迟应用）中非常重要（如果不是必需）的组件。然而，由于磁盘I/O、字符串格式化缓慢等问题，传统的日志记录机制（如写入磁盘上的日志文件）并不适用于低延迟应用。为了构建这个组件，我们使用了之前构建的多线程机制以及无锁FIFO队列。\n",
    "\n",
    "最后，我们深入讨论了网络栈的设计，包括如何创建网络套接字、如何使用它们创建TCP服务器和客户端，以及如何使用它们发布和接收多播流量。我们目前还没有使用这最后一个组件，但在后续章节中，我们将使用它来实现电子交易平台与不同市场参与者之间的通信。\n",
    "\n",
    "现在，我们将进入一个案例研究项目，在本书的后续部分，我们将构建这个项目——我们的电子交易生态系统。在下一章中，我们将首先关注系统中各个组件的高级设计，并理解它们的设计目的、设计选择背后的动机，以及系统中的信息流是如何发生的。下一章我们还将设计高级C++接口，在本书的其余部分我们将实现这些接口。"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "C++17",
   "language": "C++17",
   "name": "xcpp17"
  },
  "language_info": {
   "codemirror_mode": "text/x-c++src",
   "file_extension": ".cpp",
   "mimetype": "text/x-c++src",
   "name": "C++17",
   "version": "17"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
